Ingest Streaming Pipeline

Metadata

• System type: flow

System Intent

• What this is: The end-to-end pipeline by which a recording session — from any of three capture sources (wearable glasses, glasses-audio-only, or phone microphone) — is broken into 30-second windowed chunks on-device, uploaded to S3, and handed off to the per-window ingest Lambda. All three capture paths share the same server-side pipeline; differences are confined to client-side capture mechanics and the captureMode flag set at session start.

Mermaid Diagram

flowchart TD
  subgraph "Client — Glasses (audio+video)"
    GAudio[WearablesModule\nstartAudioCapture\n30s WAV chunks\nonAudioChunkReady event]
    GFrames[WearablesModule\nstartRecordingFrames\nevery frame → JPEG on disk\nstartFramePolling every 500ms]
  end

  subgraph "Client — Glasses-Audio (audio only)"
    GAAudio[WearablesModule\nstartAudioCapture\n30s WAV chunks\nonAudioChunkReady event]
  end

  subgraph "Client — Phone (audio only)"
    PAudio[phone-audio-capture.ts\nexpo-av\n30s WAV chunks\nonChunkReady callback]
  end

  subgraph "Client — PersistentUploadQueue"
    Queue[enqueue audio / frame]
  end

  GAudio -->|windowIndex=n, sizeBytes| Queue
  GAAudio -->|windowIndex=n, sizeBytes| Queue
  PAudio -->|windowIndex=n, sizeBytes| Queue
  GFrames -->|image/jpeg| Queue

  subgraph "Server — Audio path"
    AudioGrant["POST /sessions/{id}/audio?windowIndex=n\n→ presigned S3 PUT URL (5min TTL)"]
    AudioS3[Client PUT audio/wav → S3\nsessions/{id}/window_NNN/audio.wav]
    AudioComplete[AudioUploadCompleteFunction\nEventBridge S3 ObjectCreated]
  end

  subgraph "Server — Frame path (Android/POST)"
    FramePost["POST /sessions/{id}/frames\nframes/app.py\nDynamoDB frameCount++\nS3 write frame_NNN.jpg"]
  end

  subgraph "Server — Frame path (iOS/presigned)"
    FramePresign["GET /sessions/{id}/frames/presigned-url\n→ presigned S3 PUT URL (1h TTL)"]
    FrameS3["Client PUT image/jpeg → S3\n(URLSession background upload)"]
  end

  Queue -->|audio| AudioGrant
  AudioGrant --> AudioS3
  AudioS3 -->|ObjectCreated event| AudioComplete

  Queue -->|frame, iOS| FramePresign
  FramePresign --> FrameS3

  Queue -->|frame, Android| FramePost

  AudioComplete -->|captureMode=audio_only OR window in completedFrameWindows| IngestTrigger
  FramePost -->|60th frame AND window in completedAudioWindows| IngestTrigger

  IngestTrigger[IngestWindowFunction\nasync Lambda invoke]

Flows

Flow: glasses.audioChunks

• Core files:
• Android: main/app/wearables-module/android/src/main/java/expo/modules/wearablesmodule/WearablesModule.kt
• iOS: main/app/wearables-module/ios/WearablesModule.swift
• Client orchestration: main/app/lib/capture-session.ts

Types

AudioChunkEvent {
  filePath: string      (filesystem path to WAV file — no file:// prefix on Android, native path on iOS)
  windowIndex: number   (0-based, monotonically increasing)
  durationMs: number    (actual duration of this chunk)
  sizeBytes: number     (file size in bytes)
}

Paths

Pseudocode

// Shared across Android and iOS
pcmBytesPerChunk = 960_000  // 30s × 16kHz × 16-bit × mono
sampleRate = 16_000 Hz, mono, 16-bit

onPCMData(buffer):
  audioChunkBuffer += buffer
  if audioChunkBuffer.size >= pcmBytesPerChunk:
    flushAudioChunk(isFinal=false)

stopAudioCapture():
  flushAudioChunk(isFinal=true)   // emits partial chunk even if < 30s
  emit onAudioChunkReady

// Android: AudioRecord(MediaRecorder.AudioSource.MIC) — captures ambient room audio
// iOS: AVAudioEngine tap on HFP input port matching "ray-ban"/"meta" by name
// Both write WAV header + PCM to temp file, emit onAudioChunkReady event

Flow: glasses.frameCapture

• Core files:
• Android: main/app/wearables-module/android/src/main/java/expo/modules/wearablesmodule/WearablesModule.kt
• iOS: main/app/wearables-module/ios/WearablesModule.swift
• Polling: main/app/lib/capture-session.ts → startFramePolling

Paths

Pseudocode

// Frame write (native, event-driven)
videoFramePublisher.listen { frame ->
  jpegData = frame.makeUIImage().jpegData(compressionQuality: 0.7)
  write to framesDirectory/frame_NNNNN_<timestamp>.jpg
  frameCount++
}

// Frame poll (JS, 500ms interval)
setInterval(() => {
  files = readDir(recordingDir)
  unseen = files.filter(f => !uploadedSet.has(f))
  unseen.forEach(f => uploadQueue.enqueue("frame", f))
}, 500)

Flow: phone.audioChunks

• Core files: main/app/lib/phone-audio-capture.ts

Types

AudioChunkEvent {
  filePath: string      (absolute file:// URI to WAV file on device)
  windowIndex: number   (0-based sequential chunk index)
  durationMs: number    (actual duration of this chunk in milliseconds)
  sizeBytes: number     (file size in bytes)
}

Paths

Pseudocode

// Phone path — pure JS using expo-av
chunkTimer = setInterval(() => {
  snap currentRecording + windowIndex
  startNewChunk() → nextRecording
  windowIndex++
  finalizeChunk(currentRecording)  // emits onChunkReady
}, 30_000)

// HIGH_QUALITY preset: 44.1kHz mono WAV (compatible with /sessions/{id}/audio endpoint)
// Note: server transcription uses Groq whisper-large-v3, which handles both 44.1kHz and 16kHz

Flow: audioUpload

• Core files:
• Client: main/app/lib/capture-session.ts → uploadAudioItem
• Server (grant): main/server/api/sessions/audio/app.py
• Server (complete): main/server/events/audio_upload_complete/app.py

Types

AudioUploadGrantRequest {
  POST /sessions/{sessionId}/audio?windowIndex=n
  body: { sizeBytes: number }
}

AudioUploadGrantResponse {
  url: string           (presigned S3 PUT URL, TTL=300s)
  s3Key: string         (sessions/{sessionId}/window_NNN/audio.wav)
  windowIndex: number
  expiresIn: number     (300)
}

AudioUploadCompleteEvent (EventBridge) {
  source: "aws.s3"
  detail-type: "Object Created"
  detail.object.key: "sessions/{sessionId}/window_NNN/audio.wav"
}

Paths

Flow: frameUpload.android

• Core files: main/server/api/sessions/frames/app.py

Types

FrameUploadRequest {
  POST /sessions/{sessionId}/frames
  body: image/jpeg bytes
}

FrameUploadResponse {
  frameCount: number
  windowIndex: number
  ingestTriggered: boolean
}

Paths

Flow: frameUpload.ios

• Core files: main/app/lib/capture-session.ts → uploadItem
• Note: The iOS frame presigned URL grant is handled client-side via capture-session.ts:117. There is no separate server-side Lambda for this grant; the frames/app.py endpoint (POST /sessions/{id}/frames) is the only server-side frames handler and is used by the Android path. The iOS presigned URL flow has no corresponding server-side implementation in the current codebase.

Paths

Flow: ingestTrigger

• Core files: main/server/worldmm/pipeline/ingest_window.py
• See also: docs/docs/ingest-window.md

Paths

Key Constants

Per-Source Differences

Shared Pipeline Architecture

All three paths converge at PersistentUploadQueue on the client and at IngestWindowFunction on the server. The shared structure is:

1. Client: Audio chunks (all three) are delivered to PersistentUploadQueue.enqueue("audio", ...) with windowIndex. Frames (glasses only) are enqueued as "frame" items.
2. Client: uploadAudioItem issues a presigned URL grant (POST /sessions/{id}/audio?windowIndex=n) then PUTs audio directly to S3.
3. Server: AudioUploadCompleteFunction is triggered by EventBridge on every sessions/*/window_*/audio.wav S3 ObjectCreated event. It updates completedAudioWindows in DynamoDB and conditionally triggers ingest.
4. Server: Frame upload (POST /sessions/{id}/frames or iOS presigned PUT) updates completedFrameWindows. On the 60th frame per window, ingest is conditionally triggered.
5. Server: Ingest is deduped via a conditional ADD to ingestTriggeredWindows in DynamoDB — only one path wins per window regardless of which arrives first.
6. Server: IngestWindowFunction processes each window: loads audio from window_NNN/audio.wav, loads frames from window_NNN/frame_*.jpg. For audio_only sessions (phone or glasses-audio), frameCount=0 is passed and the Lambda skips the GPU path.

Logs

Deployment

• Mechanism: SAM (server), local only (client app)
• Deploy command:

  cd main/server && sam build && sam deploy --template .aws-sam/build/template.yaml
  

• Notes:
• AudioUploadCompleteFunction is triggered by EventBridge S3 ObjectCreated on encache-raw-memory, key prefix sessions/, suffix audio.wav.
• The old audio upload path (POST body to Lambda) is replaced by the presigned URL grant + S3 direct PUT flow. The upload-audio.md doc describes the former design; the current server code at main/server/api/sessions/audio/app.py issues presigned URLs instead of accepting audio bodies.
• iOS frame uploads use background URLSession with URLSessionConfiguration.background(withIdentifier: "com.encache.frame-uploads"); presigned URL TTL is 3600s (known limitation — should be >= 86400s).
• Android frame uploads and phone audio uploads use the fetch API path via PersistentUploadQueue.