import http from "node:http";
import { randomUUID } from "node:crypto";
import { Client } from "pg";
import gracefulShutdown from "http-graceful-shutdown";
import { EventProcessor } from "txob";
import { createProcessorClient } from "txob/pg";

const eventTypes = {
  UserInvited: "UserInvited",
  // other event types
} as const;

type EventType = keyof typeof eventTypes;

const client = new Client({
  user: process.env.POSTGRES_USER,
  password: process.env.POSTGRES_PASSWORD,
  database: process.env.POSTGRES_DB,
});
await client.connect();

const HTTP_PORT = process.env.PORT || 3000;

const processor = EventProcessor(
  createProcessorClient<EventType>(client),
  {
    UserInvited: {
      sendEmail: async (event, { signal }) => {
        // find user by event.data.userId to use relevant user data in email sending

        // email sending logic

        // use the AbortSignal `signal` to perform quick cleanup
        // during graceful shutdown enabling the processor to
        // save handler result updates to the event ASAP
      },
      publish: async (event) => {
        // publish to event bus
      },
      // other handler that should be executed when a `UserInvited` event is saved
    },
    // other event types
  }
)
processor.start();

const server = http.createServer(async (req, res) => {
  if (req.url  !== "/invite") return;

  // invite user endpoint

  const correlationId = randomUUID(); // or some value on the incoming request such as a request id

  try {
    await client.query("BEGIN");

    const userId = randomUUID();
    // save user with userId
    await client.query(`INSERT INTO users (id, email) VALUES ($1, $2)`, [userId, req.body.email]);

    // save event to `events` table
    await client.query(
      `INSERT INTO events (id, type, data, correlation_id) VALUES ($1, $2, $3, $4)`,
      [
        randomUUID(),
        eventTypes.UserInvited,
        { userId }, // other relevant data
        correlationId,
      ],
    );

    await client.query("COMMIT");
  } catch (error) {
    await client.query("ROLLBACK").catch(() => {});
  }
}).listen(HTTP_PORT, () => console.log(`listening on port ${HTTP_PORT}`));

gracefulShutdown(server, {
  onShutdown: async () => {
    // allow any actively running event handlers to finish
    // and the event processor to save the results
    await processor.stop();
  }
});