import { eq } from 'drizzle-orm'
import {
  integer,
  jsonb,
  pgTable,
  text,
  timestamp,
  uuid,
} from 'drizzle-orm/pg-core'
import type { Querier } from '@platform-modules/db'
import type { JobRegistry, DispatchOpts } from './index.js'
import type { QueueBinding } from './cf-queues.js'

export const outboxTable = pgTable('outbox', {
  id: uuid('id').primaryKey().defaultRandom(),
  aggregateType: text('aggregate_type').notNull(),
  aggregateId: text('aggregate_id').notNull(),
  eventType: text('event_type').notNull(),
  payload: jsonb('payload').notNull(),
  processedAt: timestamp('processed_at', { withTimezone: true }),
  failedAt: timestamp('failed_at', { withTimezone: true }),
  retryCount: integer('retry_count').notNull().default(0),
  lastError: text('last_error'),
  createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(),
})

export type OutboxRow = typeof outboxTable.$inferSelect
export type NewOutboxRow = typeof outboxTable.$inferInsert

const schema = { outbox: outboxTable }
type OutboxSchema = typeof schema

export async function insertOutboxRow<S extends OutboxSchema>(
  db: Querier<S>,
  values: NewOutboxRow,
): Promise<OutboxRow> {
  const [row] = await db.insert(outboxTable).values(values).returning()
  if (!row) throw new Error('insertOutboxRow: insert returned no row')
  return row
}

/** Post-commit, fire-and-forget — never throws. Sends `{ outboxId }` only. */
export function enqueueOutbox(
  queue: QueueBinding<{ outboxId: string }>,
  outboxId: string,
): void {
  void queue.send({ outboxId }).catch(() => {})
}

export async function dispatchOutboxRow<E, S extends OutboxSchema>(
  registry: JobRegistry<E>,
  db: Querier<S>,
  row: OutboxRow,
  env: E,
  opts?: DispatchOpts<E>,
): Promise<'dispatched' | 'skipped'> {
  if (row.processedAt != null) return 'skipped'

  const current = await db
    .select({ processedAt: outboxTable.processedAt })
    .from(outboxTable)
    .where(eq(outboxTable.id, row.id))
    .limit(1)
  if (current[0]?.processedAt != null) return 'skipped'

  await registry.dispatch(
    env,
    { type: row.eventType, payload: row.payload },
    opts,
  )

  await db
    .update(outboxTable)
    .set({ processedAt: new Date() })
    .where(eq(outboxTable.id, row.id))

  return 'dispatched'
}

export { schema as outboxSchema }