F-12 Media storage and processing

Brick F of the Enterprise AI Platform. Technology: Asynchronous media workers (NATS JetStream), S3 compatible object storage (SeaweedFS, Ceph RGW or cloud) with SSE-KMS through OpenBao, ClamAV scan. French version: ../../fr/bricks/F-12-media-processing.md.

Role and responsibilities

• Single entry point for every media: upload, antivirus scan, MIME sniffing, metadata extraction, classification, encryption, retention.
• Shared job framework for F-02..F-11 (queue, retries, progress, cancellation).

Position and interactions

• Execution path: event and batch.
• Emits media.uploaded, media.processed, media.deleted events; retention policies of brick E apply.

Open source and developed parts

• Open source reused: SeaweedFS or S3, ClamAV, NATS JetStream, libmagic, exiftool.
• Developed in the platform: Media service, job framework, Multimodal Contract mapping, signed download URLs.

Data model

Control Plane tables (schema eai): eai.media, eai.multimodal_jobs, eai.usage_events.

Every tenant table has tenant_id uuid NOT NULL and a row level security policy on current_setting('app.tenant_id', true)::uuid. ClickHouse usage_events is the analytics replica of eai.usage_events (ReplacingMergeTree keyed by event_id).

Contract: contracts/v1/multimodal-contract.schema.json (schema_version 1.0). Usage is always reported with contracts/v1/usage-event.schema.json, audit with contracts/v1/audit-event.schema.json.

API endpoints

Method	Path
POST	/v1/media
GET	/v1/media/{id}
DELETE	/v1/media/{id}
GET	/v1/multimodal/jobs

POST /v1/media

Request:

{
  "filename": "invoice-scan.png",
  "mime_type": "image/png",
  "operation": "vision_understanding",
  "classification": "internal",
  "content_base64": "iVBORw0KGgo..."
}

Response:

{
  "media_id": "med_01HZX3K9",
  "media_type": "image",
  "mime_type": "image/png",
  "size_bytes": 482113,
  "resolution": "1654x2339",
  "classification": "internal",
  "dlp": {
    "status": "clean",
    "findings": 0
  },
  "storage": {
    "uri": "s3://eai-media/tenant/med_01HZX3K9",
    "encrypted": true
  }
}

GET /v1/media/{id}

Response:

{
  "media_id": "med_01HZX3K9",
  "media_type": "image",
  "operation": "upload",
  "mime_type": "image/png",
  "classification": "internal",
  "created_at": "2026-10-29T10:00:00Z"
}

DELETE /v1/media/{id}

Response:

{
  "deleted": true,
  "media_id": "med_01HZX3K9"
}

GET /v1/multimodal/jobs

Response:

{
  "items": [
    {
      "job_id": "mmj_V01",
      "operation": "video_process",
      "status": "running",
      "progress": 0.4
    }
  ],
  "next_cursor": null
}

SDK examples

Python

from enterprise_ai import EnterpriseAI

client = EnterpriseAI(base_url="<control-plane-url>", api_key="eai_live_xxx", tenant="acme", locale="en")
client.multimodal.upload(open("invoice-scan.png", "rb").read(), "invoice-scan.png", "image/png", classification="internal")
client.multimodal.get_media(media_id="med_01HZX3K9")
client.multimodal.delete_media(media_id="med_01HZX3K9")
client.multimodal.jobs(status="running", limit=50)

TypeScript

import { EnterpriseAI } from "@enterprise-ai/sdk";

const client = new EnterpriseAI({ baseUrl: "<control-plane-url>", apiKey: process.env.EAI_API_KEY, tenant: "acme", locale: "en" });
await client.media.upload({ file: new Blob([await readFile("invoice-scan.png")], { type: "image/png" }), filename: "invoice-scan.png", operation: "vision_understanding", classification: "internal" });
await client.media.get("med_01HZX3K9");
await client.media.delete("med_01HZX3K9");
await client.multimodal.listJobs({ status: "running", limit: 50 });

Java

EnterpriseAIClient client = EnterpriseAIClient.builder()
    .baseUrl("<control-plane-url>")
    .apiKey(System.getenv("EAI_API_KEY"))
    .tenant("acme")
    .locale(Locale.ENGLISH)
    .build();
client.media().upload(Map.of("content_base64", Base64.getEncoder().encodeToString(Files.readAllBytes(Path.of("invoice-scan.png"))), "mime_type", "image/png", "filename", "invoice-scan.png", "classification", "internal"));
client.media().get("med_01HZX3K9");
client.media().delete("med_01HZX3K9");
client.multimodal().jobs(Map.of("status", "running", "limit", 50));

Go

client, err := eai.NewClient(eai.WithBaseURL("<control-plane-url>"), eai.WithAPIKey(os.Getenv("EAI_API_KEY")), eai.WithTenant("acme"), eai.WithLocale("en"))
if err != nil { log.Fatal(err) }
ctx := context.Background()
_, err = client.Media.Upload(ctx, map[string]any{"content_base64": base64.StdEncoding.EncodeToString(scan), "mime_type": "image/png", "filename": "invoice-scan.png", "classification": "internal"})
_, err = client.Media.Get(ctx, "med_01HZX3K9")
err = client.Media.Delete(ctx, "med_01HZX3K9")
_, err = client.Multimodal.Jobs(ctx, eai.WithQuery(map[string]string{"status": "running", "limit": "50"}))

// worker package: run a media worker consuming multimodal jobs
w, err := worker.New(worker.Config{Source: jobs, Handler: handle, Client: client, Concurrency: 2})
if err != nil { log.Fatal(err) }
err = w.Run(ctx)

Configuration

Environment variables: EAI_S3_ENDPOINT, EAI_S3_BUCKET, EAI_MEDIA_MAX_BYTES, EAI_NATS_URL.

Helm values keys (deploy/helm/enterprise-ai/values.yaml): controlPlane.config.EAI_S3_ENDPOINT, workers.enabled, workers.items.

Security

• EXIF and GPS metadata stripped by default; file type checked by content, not extension.
• Per tenant prefix and key; RLS on eai.media.
• Tenant resolved from the credential, never from the body; PostgreSQL row level security on every tenant table.
• Every mutation and policy decision writes a hash chained audit event (G-11).
• Errors use the standard envelope with message_i18n en and fr.

Observability

• Queue lag per stream, job success rate, storage growth per tenant.
• W3C traceparent propagated, X-Request-Id returned, spans carry eai.tenant_id, eai.project, eai.agent_id, eai.run_id.

FinOps units

• storage_gb_hours and bytes_transferred per media, allocated to the tenant and project.

Failure modes and resilience

• Antivirus positive -> media quarantined, incident created, upload returns validation_error.
• Object storage unavailable -> uploads rejected with 503 and Retry-After, no partial rows.

Standalone adoption

Every other F brick depends on it; it can also serve as a compliant media vault alone.

Acceptance tests

1. EICAR test file is quarantined.
2. DELETE /v1/media/{id} removes the object and writes an audit event; legal hold blocks deletion.

Implementation roadmap

Kit L4 voice evidence

Video frames extracted by the real ffmpeg then analysed by the vision adapter against a fake VLM (tests/test_multimodal_video_vision.py, MULT-007; real VLM blocked on GPU and weights). See the voice runtime guide.