import { and, desc, eq, lt, or, sql } from 'drizzle-orm';
import { bigint, integer, pgTable, text, timestamp } from 'drizzle-orm/pg-core';
import type { Querier, TransactionalDatabase } from '@platform-modules/db';
import type { D1Client } from '@platform-modules/db/sqlite/d1';

export class MediaValidationError extends Error {
  readonly name = 'MediaValidationError';
  constructor(readonly field: string, readonly detail: string) {
    super(`media validation failed: ${field} — ${detail}`);
  }
}

export class QuotaExceededError extends Error {
  readonly name = 'QuotaExceededError';
  constructor(message = 'media quota exceeded') {
    super(message);
  }
}

/** Idempotent DDL for media_assets — authoritative over media-schema.sql fixture. */
export const mediaTableSql = (): string =>
  `
CREATE TABLE IF NOT EXISTS media_assets (
  key text PRIMARY KEY,
  url text NOT NULL,
  content_type text NOT NULL,
  size bigint NOT NULL,
  width integer,
  height integer,
  uploader_id text NOT NULL,
  created_at timestamptz(3) NOT NULL DEFAULT now()
);
`.trim();

export const mediaAssets = pgTable('media_assets', {
  key: text('key').primaryKey(),
  url: text('url').notNull(),
  contentType: text('content_type').notNull(),
  size: bigint('size', { mode: 'number' }).notNull(),
  width: integer('width'),
  height: integer('height'),
  uploaderId: text('uploader_id').notNull(),
  createdAt: timestamp('created_at', { withTimezone: true, precision: 3 }).notNull().$defaultFn(() => new Date()),
});

export const mediaSchema = { mediaAssets };
export type MediaSchema = typeof mediaSchema;

export interface MediaRow {
  key: string;
  url: string;
  contentType: string;
  size: number;
  width: number | null;
  height: number | null;
  uploaderId: string;
  createdAt: Date;
}

export interface NewMedia {
  key: string;
  url: string;
  contentType: string;
  size: number;
  width?: number;
  height?: number;
  uploaderId: string;
}

/** Sum of all stored bytes — the quota denominator, derived from rows (no drift). */
export async function usedBytes(db: Querier<MediaSchema>): Promise<number> {
  const [row] = await db
    .select({ total: sql<string>`coalesce(sum(${mediaAssets.size}), 0)` })
    .from(mediaAssets);
  return Number(row?.total ?? 0);
}

export async function insertMedia(db: Querier<MediaSchema>, m: NewMedia): Promise<MediaRow> {
  const [row] = await db
    .insert(mediaAssets)
    .values({
      key: m.key,
      url: m.url,
      contentType: m.contentType,
      size: m.size,
      width: m.width ?? null,
      height: m.height ?? null,
      uploaderId: m.uploaderId,
    })
    .returning();
  return row as MediaRow;
}

/**
 * Transactional quota reserve + insert. Locks media rows, re-sums usage, inserts if within quota.
 * Caller must run storage.put BEFORE this (object outside tx); on QuotaExceededError, delete the object.
 */
export async function reserveAndInsertMedia(
  db: TransactionalDatabase<MediaSchema>,
  m: NewMedia,
  quota: number,
): Promise<MediaRow> {
  return db.transaction(async (tx) => {
    await tx.execute(sql`SELECT key FROM media_assets FOR UPDATE`);
    const [sumRow] = await tx
      .select({ total: sql<string>`coalesce(sum(${mediaAssets.size}), 0)` })
      .from(mediaAssets);
    const used = Number(sumRow?.total ?? 0);
    if (used + m.size > quota) {
      throw new QuotaExceededError();
    }
    const [row] = await tx
      .insert(mediaAssets)
      .values({
        key: m.key,
        url: m.url,
        contentType: m.contentType,
        size: m.size,
        width: m.width ?? null,
        height: m.height ?? null,
        uploaderId: m.uploaderId,
      })
      .returning();
    return row as MediaRow;
  });
}


/**
 * D1 equivalent of the Postgres quota reservation. The quota predicate and insert are one SQL statement
 * submitted through exactly one real binding batch; no callback transaction or synthetic identity exists.
 */
export async function reserveAndInsertMediaD1(
  db: D1Client<MediaSchema>,
  m: NewMedia,
  quota: number,
): Promise<MediaRow> {
  const prepared = db.prepare<{ key: string; url: string; contentType: string; size: number; width: number | null; height: number | null; uploaderId: string; createdAt: string }>(sql`
    INSERT INTO media_assets (key, url, content_type, size, width, height, uploader_id)
    SELECT ${m.key}, ${m.url}, ${m.contentType}, ${m.size}, ${m.width ?? null}, ${m.height ?? null}, ${m.uploaderId}
    WHERE (SELECT coalesce(sum(size), 0) FROM media_assets) + ${m.size} <= ${quota}
    RETURNING key, url, content_type AS "contentType", size, width, height, uploader_id AS "uploaderId", created_at AS "createdAt"
  `);
  const [result] = await db.batch([prepared]);
  const row = Array.isArray(result) ? result[0] : undefined;
  if (!row) throw new QuotaExceededError();
  return {
    ...row,
    size: Number(row.size),
    width: row.width == null ? null : Number(row.width),
    height: row.height == null ? null : Number(row.height),
    createdAt: row.createdAt instanceof Date ? row.createdAt : new Date(row.createdAt),
  } as MediaRow;
}

/**
 * Opaque cursor: `<createdAt ISO>|<key>`. Split on the FIRST separator, never the last: an ISO timestamp
 * cannot contain `|` but a key can, so `lastIndexOf` would fold part of the key into the timestamp and
 * decode to an Invalid Date. Legacy timestamp-only cursors decode with an empty key, which yields the
 * pre-composite behaviour rather than throwing.
 */
function decodeCursor(cursor: string): { createdAt: Date; key: string } {
  const sep = cursor.indexOf('|');
  const iso = sep === -1 ? cursor : cursor.slice(0, sep);
  return { createdAt: new Date(iso), key: sep === -1 ? '' : cursor.slice(sep + 1) };
}

/**
 * Keyset page newest-first by `(createdAt, key)`. The cursor carries both halves because `createdAt` has 3 ms
 * precision: rows written inside one millisecond share a timestamp, and a `createdAt`-only tie-break makes the
 * strict `<` comparison skip every same-millisecond row at a page boundary — silently dropping media from the
 * listing. `key` is unique, so the composite comparison is total and no row can fall between two pages.
 */
export async function listMedia(
  db: Querier<MediaSchema>,
  opts?: { limit?: number; cursor?: string },
): Promise<{ rows: MediaRow[]; cursor?: string }> {
  const limit = Math.min(Math.max(opts?.limit ?? 24, 1), 100);
  const after = opts?.cursor ? decodeCursor(opts.cursor) : undefined;
  const where = after
    ? or(
        lt(mediaAssets.createdAt, after.createdAt),
        and(eq(mediaAssets.createdAt, after.createdAt), lt(mediaAssets.key, after.key)),
      )
    : undefined;
  const rows = (await db
    .select()
    .from(mediaAssets)
    .where(where)
    .orderBy(desc(mediaAssets.createdAt), desc(mediaAssets.key))
    .limit(limit + 1)) as MediaRow[];
  const hasMore = rows.length > limit;
  const page = hasMore ? rows.slice(0, limit) : rows;
  const last = page[page.length - 1];
  const next = hasMore && last ? `${last.createdAt.toISOString()}|${last.key}` : undefined;
  return { rows: page, cursor: next };
}

export async function getMedia(db: Querier<MediaSchema>, key: string): Promise<MediaRow | null> {
  const [row] = await db.select().from(mediaAssets).where(eq(mediaAssets.key, key)).limit(1);
  return (row as MediaRow) ?? null;
}

/**
 * Delete a media row by key. Returns the deleted row, or null if absent (idempotent).
 * Media is a shared editorial library — tier gate only (editors upload/browse; admins delete).
 * No per-row uploaderId ACL: `uploader_id` is attribution/audit, never gated on.
 */
export async function deleteMedia(db: Querier<MediaSchema>, key: string): Promise<MediaRow | null> {
  const [row] = await db.delete(mediaAssets).where(eq(mediaAssets.key, key)).returning();
  return (row as MediaRow) ?? null;
}