import { query } from './index.js';
import type { ProviderSlug } from '../types/integration.js';

/**
 * Returns the new row id, or null when this delivery was already recorded.
 * Providers deliver at least once, so the unique constraint firing is the
 * expected path for a redelivery rather than an error.
 */
export const recordDelivery = async (input: {
  provider: ProviderSlug;
  externalId: string;
  integrationId?: number;
}): Promise<number | null> => {
  const { rows } = await query<{ id: number }>(
    `INSERT INTO ingest_events (provider, external_id, integration_id)
     VALUES ($1, $2, $3)
     ON CONFLICT (provider, external_id) DO NOTHING
     RETURNING id`,
    [input.provider, input.externalId, input.integrationId ?? null]
  );
  return rows[0]?.id ?? null;
};

export const markProcessing = async (id: number): Promise<void> => {
  await query(
    `UPDATE ingest_events
        SET status = 'processing', attempts = attempts + 1, updated_at = NOW()
      WHERE id = $1`,
    [id]
  );
};

export const markDone = async (id: number): Promise<void> => {
  await query(
    `UPDATE ingest_events
        SET status = 'done', last_error = NULL, updated_at = NOW()
      WHERE id = $1`,
    [id]
  );
};

export const markFailed = async (id: number, error: string): Promise<void> => {
  await query(
    `UPDATE ingest_events
        SET status = 'failed', last_error = $2, updated_at = NOW()
      WHERE id = $1`,
    [id, error.slice(0, 2000)]
  );
};

/**
 * The queue lives in process memory, so a restart strands anything that was
 * mid-flight. Returns rows to 'pending' so they can be picked up again.
 */
export const resetStaleProcessing = async (olderThanMs: number): Promise<number> => {
  const { rowCount } = await query(
    `UPDATE ingest_events
        SET status = 'pending', updated_at = NOW()
      WHERE status = 'processing'
        AND updated_at < NOW() - ($1 || ' milliseconds')::interval`,
    [String(olderThanMs)]
  );
  return rowCount ?? 0;
};