query-orchestration/internal/database/queries/batch.sql

-- name: CreateBatchUpload :one
INSERT INTO batch_uploads (client_id, original_filename, total_documents)
VALUES ($1, $2, $3) RETURNING id;

-- name: CreateBatchUploadWithStorage :one
INSERT INTO batch_uploads (
    client_id, 
    original_filename, 
    total_documents, 
    status,
    archive_bucket, 
    archive_key,
    file_size_bytes
) VALUES ($1, $2, $3, $4::batch_status, $5, $6, $7) 
RETURNING id, created_at;

-- name: GetBatchUpload :one
SELECT id, client_id, original_filename, total_documents, processed_documents, 
       failed_documents, invalid_type_documents, status, progress_percent,
       failed_filenames, created_at, completed_at
FROM batch_uploads 
WHERE id = $1 AND client_id = $2;

-- name: GetBatchUploadWithStorage :one
SELECT id, client_id, original_filename, total_documents, 
       processed_documents, failed_documents, invalid_type_documents,
       status, archive_bucket, archive_key, file_size_bytes,
       failed_filenames, created_at, completed_at
FROM batch_uploads 
WHERE id = $1 AND client_id = $2;

-- name: ListBatchUploads :many
SELECT id, client_id, original_filename, total_documents, processed_documents, 
       failed_documents, invalid_type_documents, status, progress_percent,
       created_at, completed_at
FROM batch_uploads 
WHERE client_id = $1
ORDER BY created_at DESC
LIMIT $2 OFFSET $3;

-- name: SetBatchTotalDocuments :exec
UPDATE batch_uploads SET total_documents = $2 WHERE id = $1;

-- name: UpdateBatchProgress :exec
UPDATE batch_uploads
SET processed_documents = $2,
    failed_documents = $3,
    invalid_type_documents = $4,
    progress_percent = $5
WHERE id = $1;

-- name: UpdateBatchStatus :exec
UPDATE batch_uploads 
SET status = $2::batch_status,
    completed_at = CASE WHEN $2::batch_status IN ('completed', 'failed', 'cancelled') THEN NOW() ELSE NULL END
WHERE id = $1;

-- name: SetBatchCompletedAt :exec
UPDATE batch_uploads
SET completed_at = @completed_at
WHERE id = @id;

-- name: AddFailedFilename :exec
UPDATE batch_uploads 
SET failed_filenames = failed_filenames || jsonb_build_array($2::text)
WHERE id = $1;

-- name: GetDocumentsByBatchId :many
SELECT id, clientId, hash 
FROM documents 
WHERE batch_id = $1;

-- name: CountDocumentsByBatchId :one
SELECT COUNT(*) as count
FROM documents 
WHERE batch_id = $1;

-- name: GetUnprocessedBatches :many
SELECT id, client_id, original_filename, total_documents,
       processed_documents, failed_documents, invalid_type_documents,
       status, progress_percent, created_at, completed_at
FROM batch_uploads
WHERE status = 'processing'
ORDER BY created_at ASC;

-- name: InsertBatchDocumentOutcome :exec
-- Upsert: if a row for this (batch_id, filename) already exists (e.g. batch
-- retry after partial failure), overwrite it with the latest outcome.
INSERT INTO batch_document_outcomes (batch_id, filename, outcome, error_detail, document_id)
VALUES ($1, $2, $3::batch_outcome_status, $4, $5)
ON CONFLICT (batch_id, filename) DO UPDATE
SET outcome = EXCLUDED.outcome,
    error_detail = EXCLUDED.error_detail,
    document_id = EXCLUDED.document_id,
    updated_at = NOW();

-- name: UpdateBatchDocumentOutcomeByDocumentID :many
-- Updates the outcome for a document progressing through the pipeline.
-- Only updates rows in non-terminal pipeline states to prevent overwriting
-- terminal outcomes (e.g. a 'duplicate' row from a different batch that
-- shares the same document_id).
UPDATE batch_document_outcomes
SET outcome = $2::batch_outcome_status,
    error_detail = $3,
    updated_at = NOW()
WHERE document_id = $1
  AND outcome IN ('submitted', 'init_complete', 'sync_complete')
RETURNING batch_id;

-- name: ResolveBatchDocumentOutcome :exec
-- Called by docInitRunner to set the document_id on a previously-submitted
-- outcome row and update the outcome to an init-stage result.
-- Finds the row by batch_id + filename since document_id is NULL at this point.
UPDATE batch_document_outcomes
SET document_id = $3,
    outcome = $4::batch_outcome_status,
    error_detail = $5,
    updated_at = NOW()
WHERE batch_id = $1 AND filename = $2 AND document_id IS NULL;

-- name: ListBatchDocumentOutcomes :many
-- Returns all outcome rows for a batch, with clean failure detail
-- joined from currentCleanEntries when applicable.
SELECT
    bdo.id,
    bdo.batch_id,
    bdo.filename,
    bdo.outcome,
    bdo.error_detail,
    bdo.document_id,
    bdo.created_at,
    bdo.updated_at,
    cce.fail as clean_fail
FROM batch_document_outcomes bdo
LEFT JOIN currentCleanEntries cce ON cce.documentId = bdo.document_id
WHERE bdo.batch_id = $1
ORDER BY bdo.created_at ASC;

-- name: RecordBatchFolderCandidate :exec
-- Records a non-root folder touched by a batch upload. Once a batch observes a
-- folder as newly created, retries must preserve that fact.
INSERT INTO batch_folder_candidates (batch_id, folder_id, path, existed_before)
VALUES (@batch_id, @folder_id, @path, @existed_before)
ON CONFLICT (batch_id, folder_id) DO UPDATE
SET path = EXCLUDED.path,
    existed_before = batch_folder_candidates.existed_before AND EXCLUDED.existed_before;

-- name: ListBatchFolderCandidates :many
SELECT batch_id, folder_id, path, existed_before, created_at
FROM batch_folder_candidates
WHERE batch_id = @batch_id
ORDER BY path;

-- name: LockBatchUploadForFolderCleanup :one
SELECT id, status, total_documents, created_at, completed_at, folder_cleanup_completed_at
FROM batch_uploads
WHERE id = @batch_id
FOR UPDATE;

-- name: CountBatchOutcomes :one
SELECT COUNT(*)::int
FROM batch_document_outcomes
WHERE batch_id = @batch_id;

-- name: CountNonTerminalBatchOutcomes :one
SELECT COUNT(*)::int
FROM batch_document_outcomes
WHERE batch_id = @batch_id
  AND outcome IN ('submitted', 'init_complete', 'sync_complete');

-- name: CountAcceptedBatchOutcomes :one
SELECT COUNT(*)::int
FROM batch_document_outcomes
WHERE batch_id = @batch_id
  AND outcome IN ('clean_passed', 'sync_skipped');

-- name: ListFolderCleanupReadyBatchIDs :many
-- @sqlc-vet-disable
SELECT bu.id
FROM batch_uploads bu
WHERE bu.folder_cleanup_completed_at IS NULL
  AND bu.status IN ('completed', 'failed', 'cancelled')
  AND EXISTS (
      SELECT 1
      FROM batch_folder_candidates bfc
      WHERE bfc.batch_id = bu.id
  )
  AND (
      (
          bu.total_documents > 0
          AND
          (SELECT COUNT(*) FROM batch_document_outcomes bdo WHERE bdo.batch_id = bu.id) = bu.total_documents
          AND NOT EXISTS (
              SELECT 1
              FROM batch_document_outcomes bdo
              WHERE bdo.batch_id = bu.id
                AND bdo.outcome IN ('submitted', 'init_complete', 'sync_complete')
          )
      )
      OR (
          bu.status IN ('failed', 'cancelled')
          AND (SELECT COUNT(*) FROM batch_document_outcomes bdo WHERE bdo.batch_id = bu.id) > 0
          AND bu.completed_at IS NOT NULL
          AND bu.completed_at <= NOW() - (sqlc.arg(stale_after_seconds)::int * INTERVAL '1 second')
      )
  )
ORDER BY bu.created_at ASC
LIMIT sqlc.arg(limit_count);

-- name: DetachRejectedBatchDocumentsFromCandidateFolders :execrows
-- @sqlc-vet-disable
UPDATE documents d
SET folderId = NULL
WHERE d.batch_id = @batch_id
  AND d.folderId IS NOT NULL
  AND EXISTS (
      SELECT 1
      FROM batch_folder_candidates bfc
      WHERE bfc.batch_id = @batch_id
        AND bfc.folder_id = d.folderId
  )
  AND EXISTS (
      SELECT 1
      FROM batch_document_outcomes bdo
      WHERE bdo.batch_id = @batch_id
        AND bdo.document_id = d.id
        AND (
            bdo.outcome IN (
                'duplicate',
                'init_duplicate',
                'invalid_type',
                'failed_open',
                'failed_read',
                'failed_s3_upload',
                'failed_upload',
                'clean_failed'
            )
            OR (@include_nonterminal::boolean AND bdo.outcome IN ('submitted', 'init_complete', 'sync_complete'))
        )
  );

-- name: DetachRejectedBatchDocumentUploadsFromCandidateFolders :execrows
-- @sqlc-vet-disable
UPDATE documentUploads du
SET folder_id = NULL
WHERE du.batch_id = @batch_id
  AND du.folder_id IS NOT NULL
  AND EXISTS (
      SELECT 1
      FROM batch_folder_candidates bfc
      WHERE bfc.batch_id = @batch_id
        AND bfc.folder_id = du.folder_id
  )
  AND EXISTS (
      SELECT 1
      FROM batch_document_outcomes bdo
      WHERE bdo.batch_id = @batch_id
        AND bdo.filename = du.filename
        AND (
            bdo.outcome IN (
                'duplicate',
                'init_duplicate',
                'invalid_type',
                'failed_open',
                'failed_read',
                'failed_s3_upload',
                'failed_upload',
                'clean_failed'
            )
            OR (@include_nonterminal::boolean AND bdo.outcome IN ('submitted', 'init_complete', 'sync_complete'))
        )
  );

-- name: MarkBatchFolderCleanupCompleted :exec
UPDATE batch_uploads
SET folder_cleanup_completed_at = NOW()
WHERE id = @batch_id
  AND folder_cleanup_completed_at IS NULL;