feat(ingest): accept uploads and queue them as jobs

docker-compose.publish.yml

@@ -54,6 +54,7 @@ services:
     restart: unless-stopped
     volumes:
       - staging-data:/usr/src/app/staging_data:ro
+      - fswatch-state:/var/lib/tagbase-fswatch
 
   nginx:
     image: ghcr.io/tagbase/tagbase-nginx:${TAGBASE_VERSION:-latest}
@@ -317,5 +318,6 @@ volumes:
   lego-data:
   postgis-data:
   staging-data:
+  fswatch-state:
   dbbackups:
   tagbase-logs:

docker-compose.test.yml

@@ -100,6 +100,7 @@ services:
       - TAGBASE_INGEST_BASE=tagbase_server:5433/tagbase/api/v0
     volumes:
       - ./staging_data:/usr/src/app/staging_data:ro
+      - fswatch-state:/var/lib/tagbase-fswatch
     networks:
       - test-network
 
@@ -195,6 +196,7 @@ networks:
   test-network:
 
 volumes:
+  fswatch-state:
   tagbase-test-pgdata:
   tagbase-test-alloy:
   tagbase-test-prometheus:

docker-compose.yml

@@ -59,6 +59,7 @@ services:
     restart: unless-stopped
     volumes:
       - ./staging_data:/usr/src/app/staging_data:ro
+      - fswatch-state:/var/lib/tagbase-fswatch
   nginx:
     build:
       context: ./services/nginx
@@ -375,6 +376,7 @@ networks:
   internal-network:
 
 volumes:
+  fswatch-state:
   alloy-data:
   prometheus-data:
   loki-data:

openapi.yaml

@@ -70,20 +70,24 @@ paths:
         - $ref: "#/components/parameters/version"
       requestBody:
         content:
+          application/gzip:
+            schema:
+              $ref: "#/components/schemas/SyncUpload"
           application/octet-stream:
             schema:
-              description: Compressed binary e.g. zip file containing one or more eTUFF files
-              format: binary
-              maxLength: 1000000000
-              minLength: 1
-              type: string
+              $ref: "#/components/schemas/SyncUpload"
+          application/x-gzip:
+            schema:
+              $ref: "#/components/schemas/SyncUpload"
+          application/x-tar:
+            schema:
+              $ref: "#/components/schemas/SyncUpload"
+          application/zip:
+            schema:
+              $ref: "#/components/schemas/SyncUpload"
           text/plain:
             schema:
-              description: Plain text eTUFF file
-              format: binary
-              maxLength: 1000000000
-              minLength: 1
-              type: string
+              $ref: "#/components/schemas/SyncUpload"
         required: true
         x-body-name: body
       responses:
@@ -95,13 +99,88 @@ paths:
           description: A success message confirming ingestion.
         "400":
           $ref: "#/components/responses/BadRequest"
+        "413":
+          $ref: "#/components/responses/PayloadTooLarge"
         "500":
           $ref: "#/components/responses/InternalError"
       summary: Post a local file and perform a ingest operation
       tags:
         - ingest
       x-codegen-request-body-name: body
       x-openapi-router-controller: tagbase_server.controllers.ingest_controller
+  /ingest/jobs:
+    post:
+      description: >
+        Accept a .txt, .zip, .tar.gz, or .tgz upload up to 2 GiB, store it,
+        and enqueue ingest. Poll GET /ingest/jobs/{job_id} until the job
+        succeeds or fails.
+      operationId: create_ingest_job
+      parameters:
+        - $ref: "#/components/parameters/filename"
+        - $ref: "#/components/parameters/notes"
+        - $ref: "#/components/parameters/type"
+        - $ref: "#/components/parameters/version"
+      requestBody:
+        content:
+          application/gzip:
+            schema:
+              $ref: "#/components/schemas/JobUpload"
+          application/octet-stream:
+            schema:
+              $ref: "#/components/schemas/JobUpload"
+          application/x-gzip:
+            schema:
+              $ref: "#/components/schemas/JobUpload"
+          application/x-tar:
+            schema:
+              $ref: "#/components/schemas/JobUpload"
+          application/zip:
+            schema:
+              $ref: "#/components/schemas/JobUpload"
+          text/plain:
+            schema:
+              $ref: "#/components/schemas/JobUpload"
+        required: true
+        x-body-name: body
+      responses:
+        "202":
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/IngestAccepted"
+          description: Request accepted. Ingest job enqueued
+        "400":
+          $ref: "#/components/responses/BadRequest"
+        "413":
+          $ref: "#/components/responses/PayloadTooLarge"
+        "500":
+          $ref: "#/components/responses/InternalError"
+      summary: Enqueue a large file for ingest
+      tags:
+        - ingest
+      x-codegen-request-body-name: body
+      x-openapi-router-controller: tagbase_server.controllers.ingest_controller
+  /ingest/jobs/{job_id}:
+    parameters:
+      - $ref: "#/components/parameters/jobId"
+    get:
+      description: Read the status of one ingest job.
+      operationId: get_ingest_job
+      responses:
+        "200":
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/IngestJob"
+          description: The current ingest job.
+        "404":
+          $ref: "#/components/responses/NotFound"
+        "500":
+          $ref: "#/components/responses/InternalError"
+      summary: Read one ingest job
+      tags:
+        - ingest
+      x-openapi-router-controller: tagbase_server.controllers.ingest_controller
   /tags:
     delete:
       description: Delete all tags
@@ -205,13 +284,38 @@ components:
         application/problem+json:
           schema:
             $ref: "#/components/schemas/Problem"
+    NotFound:
+      description: The requested resource does not exist.
+      content:
+        application/problem+json:
+          schema:
+            $ref: "#/components/schemas/Problem"
+    PayloadTooLarge:
+      description: The request body exceeds the limit for this operation.
+      content:
+        application/problem+json:
+          schema:
+            $ref: "#/components/schemas/Problem"
     InternalError:
       description: Unexpected server error. Contact admin detailed in openapi.yaml.
       content:
         application/problem+json:
           schema:
             $ref: "#/components/schemas/Problem"
   parameters:
+    jobId:
+      description: Ingest job id
+      explode: false
+      in: path
+      name: job_id
+      required: true
+      schema:
+        format: uuid
+        maxLength: 36
+        minLength: 36
+        pattern: ^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$
+        type: string
+      style: simple
     filename:
       description: Free-form text field to explicitly define the name of the file to be persisted
       explode: true
@@ -398,6 +502,91 @@ components:
           example: 4bf92f3577b34da6a3ce929d0e0e4736
       title: Ingest200
       type: object
+    SyncUpload:
+      description: Synchronous ingest body, at most 100 MiB
+      format: binary
+      maxLength: 104857600
+      minLength: 1
+      type: string
+    JobUpload:
+      description: Ingest job body, at most 2 GiB
+      format: binary
+      maxLength: 2147483648
+      minLength: 1
+      type: string
+    IngestAccepted:
+      description: HTTP 202 response after an ingest job is queued
+      required:
+        - id
+        - status
+      example:
+        id: 4bf92f35-77b3-4da6-a3ce-929d0e0e4736
+        status: queued
+      properties:
+        id:
+          description: Ingest job id
+          format: uuid
+          type: string
+          minLength: 36
+          maxLength: 36
+          pattern: ^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$
+          example: 4bf92f35-77b3-4da6-a3ce-929d0e0e4736
+        status:
+          description: Job status at accept time
+          type: string
+          minLength: 6
+          maxLength: 9
+          pattern: ^queued$
+          example: queued
+      title: IngestAccepted
+      type: object
+    IngestJob:
+      description: Status of one ingest job
+      required:
+        - id
+        - status
+        - files_done
+      example:
+        id: 4bf92f35-77b3-4da6-a3ce-929d0e0e4736
+        status: succeeded
+        files_done: 1
+        files_total: 1
+      properties:
+        id:
+          description: Ingest job id
+          format: uuid
+          type: string
+          minLength: 36
+          maxLength: 36
+          pattern: ^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$
+          example: 4bf92f35-77b3-4da6-a3ce-929d0e0e4736
+        status:
+          description: queued, running, succeeded, or failed
+          type: string
+          minLength: 6
+          maxLength: 9
+          pattern: ^(queued|running|succeeded|failed)$
+          example: succeeded
+        error:
+          description: Present when the job failed
+          type: string
+          minLength: 1
+          maxLength: 10000
+          pattern: ^(?!\s*$).+$
+        files_total:
+          description: eTUFF members in the upload
+          type: integer
+          format: int32
+          minimum: 0
+          maximum: 10000
+        files_done:
+          description: eTUFF members already processed
+          type: integer
+          format: int32
+          minimum: 0
+          maximum: 10000
+      title: IngestJob
+      type: object
     Metadata:
       description: Contains the ingested tag metadata consistent with the eTUFF specification
       example:

services/docker-cron/crontab

@@ -1 +1,2 @@
 */1 * * * * tagbase . /usr/src/app/.env && /usr/local/bin/python3 /usr/src/app/poll_metadata_and_obs_types.py >/proc/1/fd/1 2>/proc/1/fd/2
+15 3 * * * tagbase . /usr/src/app/.env && /usr/local/bin/python3 /usr/src/app/purge_ingest_jobs.py >/proc/1/fd/1 2>/proc/1/fd/2

services/docker-cron/entrypoint.sh

@@ -8,6 +8,7 @@ POSTGRES_PORT="${POSTGRES_PORT:-5432}"
 cat >/usr/src/app/.env <<EOF
 export POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
 export POSTGRES_PORT=${POSTGRES_PORT}
+export INGEST_JOB_RETENTION_DAYS=${INGEST_JOB_RETENTION_DAYS:-30}
 EOF
 
 echo "Seeding metadata_types and observation_types..."

services/docker-cron/purge_ingest_jobs.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python3
+"""Delete terminal ingest_job rows older than the retention window."""
+
+import os
+import sys
+
+import psycopg2
+
+DEFAULT_RETENTION_DAYS = 30
+
+
+def retention_days():
+    raw = os.getenv("INGEST_JOB_RETENTION_DAYS", str(DEFAULT_RETENTION_DAYS))
+    return int(raw)
+
+
+def connect():
+    try:
+        return psycopg2.connect(
+            dbname="tagbase",
+            user="tagbase",
+            host="postgis",
+            port=os.getenv("POSTGRES_PORT", "5432"),
+            password=os.getenv("POSTGRES_PASSWORD", "tagbase"),
+        )
+    except psycopg2.OperationalError as error:
+        print("Error connecting to DB: ", error)
+        sys.exit(1)
+
+
+def purge(connection, days):
+    with connection.cursor() as cursor:
+        cursor.execute(
+            """
+            DELETE FROM ingest_job
+            WHERE status IN ('succeeded', 'failed')
+              AND updated_at < now() - (%s * interval '1 day')
+            """,
+            (days,),
+        )
+        deleted = cursor.rowcount
+    connection.commit()
+    return deleted
+
+
+if __name__ == "__main__":
+    window = retention_days()
+    conn = connect()
+    try:
+        removed = purge(conn, window)
+    finally:
+        conn.close()
+    print(f"Purged {removed} ingest_job row(s) older than {window} day(s)")

services/fswatch/Dockerfile

@@ -7,7 +7,9 @@ RUN apt-get update \
     install -y curl fswatch iputils-ping \
     # Remove package lists for smaller image sizes
     && rm -rf /var/lib/apt/lists/* \
-    && useradd -ms /bin/bash tagbase
+    && useradd -ms /bin/bash tagbase \
+    && mkdir -p /var/lib/tagbase-fswatch \
+    && chown tagbase:tagbase /var/lib/tagbase-fswatch
 
 # Copy owned by root and grant the runtime user read+execute only, so the
 # non-root user running the container cannot modify the script it executes.