import * as SqlClient from "effect/unstable/sql/SqlClient";
import * as SqlSchema from "effect/unstable/sql/SqlSchema";
import { EventId, NonNegativeInt, ThreadId } from "@t3tools/contracts";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Schema from "effect/Schema";
import * as Struct from "effect/Struct";

import { toPersistenceDecodeError, toPersistenceSqlError } from "../Errors.ts";

import {
  DeleteProjectionThreadActivitiesInput,
  ListProjectionThreadActivitiesInput,
  ProjectionThreadActivity,
  ProjectionThreadActivityRepository,
  type ProjectionThreadActivityRepositoryShape,
} from "../Services/ProjectionThreadActivities.ts";

const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields(
  Struct.assign({
    payload: Schema.fromJsonString(Schema.Unknown),
    sequence: Schema.NullOr(NonNegativeInt),
  }),
);
type ProjectionThreadActivityDbRow = typeof ProjectionThreadActivityDbRowSchema.Type;

const fromDbRow = (row: ProjectionThreadActivityDbRow): ProjectionThreadActivity => ({
  activityId: row.activityId,
  threadId: row.threadId,
  turnId: row.turnId,
  tone: row.tone,
  kind: row.kind,
  summary: row.summary,
  payload: row.payload,
  ...(row.sequence !== null ? { sequence: row.sequence } : {}),
  createdAt: row.createdAt,
});

const PendingUserInputCountRowSchema = Schema.Struct({
  count: NonNegativeInt,
});

const ExistingUserInputActivityRowSchema = Schema.Struct({
  threadId: ThreadId,
  requestId: Schema.NullOr(Schema.String),
});

const UserInputActivityKeySchema = Schema.Struct({
  threadId: ThreadId,
  requestId: Schema.String,
});
type UserInputActivityKey = typeof UserInputActivityKeySchema.Type;

const UserInputStateRowSchema = Schema.Struct({
  state: Schema.Literals(["pending", "cleared"]),
});
type UserInputState = typeof UserInputStateRowSchema.Type.state;

function userInputStateFromActivity(activity: {
  readonly kind: string;
  readonly payload: unknown;
}): { readonly requestId: string; readonly state: UserInputState } | null {
  if (typeof activity.payload !== "object" || activity.payload === null) {
    return null;
  }
  const payload = activity.payload as Record<string, unknown>;
  if (typeof payload.requestId !== "string") {
    return null;
  }
  if (activity.kind === "user-input.requested") {
    return { requestId: payload.requestId, state: "pending" };
  }
  if (activity.kind === "user-input.resolved") {
    return { requestId: payload.requestId, state: "cleared" };
  }
  if (activity.kind !== "provider.user-input.respond.failed") {
    return null;
  }
  const detail = typeof payload.detail === "string" ? payload.detail.toLowerCase() : "";
  if (
    !detail.includes("stale pending user-input request") &&
    !detail.includes("unknown pending user-input request") &&
    !detail.includes("unknown pending user input request") &&
    !detail.includes("unknown pending codex user input request")
  ) {
    return null;
  }
  return { requestId: payload.requestId, state: "cleared" };
}

function userInputKeyId(key: UserInputActivityKey): string {
  return `${key.threadId.length}:${key.threadId}${key.requestId}`;
}

function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) {
  return (cause: unknown) =>
    Schema.isSchemaError(cause)
      ? toPersistenceDecodeError(decodeOperation)(cause)
      : toPersistenceSqlError(sqlOperation)(cause);
}

const makeProjectionThreadActivityRepository = Effect.gen(function* () {
  const sql = yield* SqlClient.SqlClient;

  const getExistingUserInputActivity = SqlSchema.findOneOption({
    Request: Schema.Struct({ activityId: EventId }),
    Result: ExistingUserInputActivityRowSchema,
    execute: ({ activityId }) =>
      sql`
        SELECT
          thread_id AS "threadId",
          user_input_request_id AS "requestId"
        FROM projection_thread_activities
        WHERE activity_id = ${activityId}
          AND user_input_state IS NOT NULL
        LIMIT 1
      `,
  });

  const getLatestUserInputState = SqlSchema.findOneOption({
    Request: UserInputActivityKeySchema,
    Result: UserInputStateRowSchema,
    execute: ({ threadId, requestId }) =>
      sql`
        SELECT user_input_state AS "state"
        FROM projection_thread_activities
        WHERE thread_id = ${threadId}
          AND user_input_request_id = ${requestId}
          AND user_input_state IS NOT NULL
        ORDER BY created_at DESC, activity_id DESC
        LIMIT 1
      `,
  });

  const upsertProjectionThreadActivityRow = SqlSchema.void({
    Request: ProjectionThreadActivity,
    execute: (row) => {
      const userInputState = userInputStateFromActivity(row);
      return sql`
            INSERT INTO projection_thread_activities (
              activity_id,
              thread_id,
              turn_id,
              tone,
              kind,
              summary,
              payload_json,
              sequence,
              created_at,
              user_input_request_id,
              user_input_state
            )
            VALUES (
              ${row.activityId},
              ${row.threadId},
              ${row.turnId},
              ${row.tone},
              ${row.kind},
              ${row.summary},
              ${JSON.stringify(row.payload)},
              ${row.sequence ?? null},
              ${row.createdAt},
              ${userInputState?.requestId ?? null},
              ${userInputState?.state ?? null}
            )
            ON CONFLICT (activity_id)
            DO UPDATE SET
              thread_id = excluded.thread_id,
              turn_id = excluded.turn_id,
              tone = excluded.tone,
              kind = excluded.kind,
              summary = excluded.summary,
              payload_json = excluded.payload_json,
              sequence = excluded.sequence,
              created_at = excluded.created_at,
              user_input_request_id = excluded.user_input_request_id,
              user_input_state = excluded.user_input_state
          `;
    },
  });

  const listProjectionThreadActivityRows = SqlSchema.findAll({
    Request: ListProjectionThreadActivitiesInput,
    Result: ProjectionThreadActivityDbRowSchema,
    execute: ({ threadId }) =>
      sql`
        SELECT
          activity_id AS "activityId",
          thread_id AS "threadId",
          turn_id AS "turnId",
          tone,
          kind,
          summary,
          payload_json AS "payload",
          sequence,
          created_at AS "createdAt"
        FROM projection_thread_activities
        WHERE thread_id = ${threadId}
        ORDER BY
          CASE WHEN sequence IS NULL THEN 0 ELSE 1 END ASC,
          sequence ASC,
          created_at ASC,
          activity_id ASC
      `,
  });

  const readPendingUserInputCount = SqlSchema.findOneOption({
    Request: ListProjectionThreadActivitiesInput,
    Result: PendingUserInputCountRowSchema,
    execute: ({ threadId }) =>
      sql`
        SELECT pending_count AS "count"
        FROM projection_thread_user_input_summaries
        WHERE thread_id = ${threadId}
        LIMIT 1
      `,
  });

  const rebuildPendingUserInputCount = SqlSchema.findOne({
    Request: ListProjectionThreadActivitiesInput,
    Result: PendingUserInputCountRowSchema,
    execute: ({ threadId }) =>
      sql`
        WITH ranked AS (
          SELECT
            user_input_state,
            ROW_NUMBER() OVER (
              PARTITION BY user_input_request_id
              ORDER BY created_at DESC, activity_id DESC
            ) AS row_number
          FROM projection_thread_activities
          WHERE thread_id = ${threadId}
            AND user_input_request_id IS NOT NULL
            AND user_input_state IS NOT NULL
        )
        SELECT COUNT(*) AS "count"
        FROM ranked
        WHERE row_number = 1
          AND user_input_state = 'pending'
      `,
  });

  const getOrRebuildPendingUserInputCount = Effect.fn(
    "ProjectionThreadActivityRepository.getOrRebuildPendingUserInputCount",
  )(function* (input: ListProjectionThreadActivitiesInput) {
    const existing = yield* readPendingUserInputCount(input);
    if (Option.isSome(existing)) {
      return existing.value.count;
    }

    const rebuilt = yield* rebuildPendingUserInputCount(input);
    yield* sql`
      INSERT INTO projection_thread_user_input_summaries (thread_id, pending_count)
      VALUES (${input.threadId}, ${rebuilt.count})
      ON CONFLICT (thread_id)
      DO UPDATE SET pending_count = excluded.pending_count
    `;
    return rebuilt.count;
  });

  const deleteProjectionThreadActivityRows = SqlSchema.void({
    Request: DeleteProjectionThreadActivitiesInput,
    execute: ({ threadId }) =>
      sql`
        DELETE FROM projection_thread_activities
        WHERE thread_id = ${threadId}
      `,
  });

  const resetPendingUserInputSummary = SqlSchema.void({
    Request: DeleteProjectionThreadActivitiesInput,
    execute: ({ threadId }) =>
      sql`
        INSERT INTO projection_thread_user_input_summaries (thread_id, pending_count)
        VALUES (${threadId}, 0)
        ON CONFLICT (thread_id)
        DO UPDATE SET pending_count = 0
      `,
  });

  const upsert: ProjectionThreadActivityRepositoryShape["upsert"] = (row) =>
    sql
      .withTransaction(
        Effect.gen(function* () {
          yield* getOrRebuildPendingUserInputCount({ threadId: row.threadId });
          const previous = yield* getExistingUserInputActivity({ activityId: row.activityId });
          if (Option.isSome(previous) && previous.value.threadId !== row.threadId) {
            yield* getOrRebuildPendingUserInputCount({ threadId: previous.value.threadId });
          }
          const next = userInputStateFromActivity(row);
          const keys = new Map<string, UserInputActivityKey>();
          if (Option.isSome(previous) && previous.value.requestId !== null) {
            const key = {
              threadId: previous.value.threadId,
              requestId: previous.value.requestId,
            };
            keys.set(userInputKeyId(key), key);
          }
          if (next !== null) {
            const key = { threadId: row.threadId, requestId: next.requestId };
            keys.set(userInputKeyId(key), key);
          }

          const before = new Map<string, boolean>();
          for (const [keyId, key] of keys) {
            const state = yield* getLatestUserInputState(key);
            before.set(keyId, Option.isSome(state) && state.value.state === "pending");
          }

          yield* upsertProjectionThreadActivityRow(row);

          const deltas = new Map<string, number>();
          for (const [keyId, key] of keys) {
            const state = yield* getLatestUserInputState(key);
            const wasPending = before.get(keyId) ?? false;
            const isPending = Option.isSome(state) && state.value.state === "pending";
            const delta = Number(isPending) - Number(wasPending);
            if (delta !== 0) {
              deltas.set(key.threadId, (deltas.get(key.threadId) ?? 0) + delta);
            }
          }
          for (const [threadId, delta] of deltas) {
            yield* sql`
              INSERT INTO projection_thread_user_input_summaries (thread_id, pending_count)
              VALUES (${threadId}, ${Math.max(delta, 0)})
              ON CONFLICT (thread_id)
              DO UPDATE SET pending_count = pending_count + ${delta}
            `;
          }

          const affectedThreadIds = new Set(Array.from(keys.values(), (key) => key.threadId));
          for (const threadId of affectedThreadIds) {
            const pendingUserInputCount = yield* getOrRebuildPendingUserInputCount({ threadId });
            yield* sql`
              UPDATE projection_threads
              SET pending_user_input_count = ${pendingUserInputCount}
              WHERE thread_id = ${threadId}
            `;
          }
        }),
      )
      .pipe(
        Effect.mapError(
          toPersistenceSqlOrDecodeError(
            "ProjectionThreadActivityRepository.upsert:query",
            "ProjectionThreadActivityRepository.upsert:encodeRequest",
          ),
        ),
      );

  const listByThreadId: ProjectionThreadActivityRepositoryShape["listByThreadId"] = (input) =>
    listProjectionThreadActivityRows(input).pipe(
      Effect.mapError(
        toPersistenceSqlOrDecodeError(
          "ProjectionThreadActivityRepository.listByThreadId:query",
          "ProjectionThreadActivityRepository.listByThreadId:decodeRows",
        ),
      ),
      Effect.map((rows) => rows.map(fromDbRow)),
    );

  const countPendingUserInputsByThreadId: ProjectionThreadActivityRepositoryShape["countPendingUserInputsByThreadId"] =
    (input) =>
      sql
        .withTransaction(getOrRebuildPendingUserInputCount(input))
        .pipe(
          Effect.mapError(
            toPersistenceSqlOrDecodeError(
              "ProjectionThreadActivityRepository.countPendingUserInputsByThreadId:query",
              "ProjectionThreadActivityRepository.countPendingUserInputsByThreadId:decodeRow",
            ),
          ),
        );

  const deleteByThreadId: ProjectionThreadActivityRepositoryShape["deleteByThreadId"] = (input) =>
    sql
      .withTransaction(
        Effect.gen(function* () {
          yield* deleteProjectionThreadActivityRows(input);
          yield* resetPendingUserInputSummary(input);
          yield* sql`
            UPDATE projection_threads
            SET pending_user_input_count = 0
            WHERE thread_id = ${input.threadId}
          `;
        }),
      )
      .pipe(
        Effect.mapError(
          toPersistenceSqlError("ProjectionThreadActivityRepository.deleteByThreadId:query"),
        ),
      );

  return {
    upsert,
    listByThreadId,
    countPendingUserInputsByThreadId,
    deleteByThreadId,
  } satisfies ProjectionThreadActivityRepositoryShape;
});

export const ProjectionThreadActivityRepositoryLive = Layer.effect(
  ProjectionThreadActivityRepository,
  makeProjectionThreadActivityRepository,
);