// The knowledge jobs and the subscribers that enqueue them.
//
// A web request writes the document row and publishes the outbox event in one transaction; the
// module's own subscriber (running in the worker's outbox relay) turns the event into the job,
// with the (collection, file) pair as the singleton key — a redelivered event adds no second job.
//
// The relay delivers every record in its organization's scope, with a system actor, and the worker binds
// `KNOWLEDGE_SCOPE` to that delivery scope. The organization comes from there, never from the payload
// (decision AT-F6); only the workspace and the project come from the record, which this module or the files
// module wrote inside that organization's transaction, as `projectScopeOf` of @conusai-kit/files does. A job,
// unlike a delivery, runs in the project scope it was enqueued in.

import { type EventContext, internal, type JobDef, serviceRefusal, subscribe } from "@conusai-kit/core"
import { DATA_SOURCE, JOB_QUEUE } from "@conusai-kit/data"
import { fileDeleted, type ProcessorJobPayload, processorJobPayloadSchema } from "@conusai-kit/files"
import { NATIVE_JOB_ADMISSION } from "@conusai-kit/platform"
import { projects, workspaces } from "@conusai-kit/tenancy"
import { z } from "zod"
import {
  AUTO_INDEX_JOB,
  documentIndexRequested,
  documentRemoveRequested,
  INDEX_JOB,
  REMOVE_JOB,
} from "../contract/index.js"
import { autoIndexUpload } from "../core/auto-index-job.js"
import { requestSourceRemoval } from "../core/documents.js"
import { indexDocument, removeDocument } from "../core/index-job.js"
import { KNOWLEDGE_SCOPE, type KnowledgeScope } from "../core/ports.js"
import { knowledgeDeps, knowledgeSearchDeps } from "./procedures.js"

export const documentPayload = z.object({
  documentId: z.string(),
  collectionName: z.string(),
  fileId: z.string(),
  /** A remove job's: the object to remove, resolved when the removal was asked for; absent only on an older record. */
  storageKey: z.string().optional(),
})
export type DocumentJobPayload = z.output<typeof documentPayload>

export const indexJobKey = (collectionName: string, fileId: string): string =>
  `knowledge.index.${collectionName}.${fileId}`
export const removeJobKey = (collectionName: string, fileId: string): string =>
  `knowledge.remove.${collectionName}.${fileId}`

/**
 * The project a delivery acts in: the delivery's organization and actor, and the record's workspace and project.
 * A record that names another organization is refused here rather than acted on there, and so is one whose workspace
 * the organization does not have or whose project is not in that workspace: both are read in the delivery's own
 * organization before anything is enqueued or changed. That refusal can never succeed on a retry; the relay has no
 * terminal-failure path yet, so it is retried until marked stuck, with the reason (a kit-jobs gap, recorded in the
 * template lessons). An archived project is acted on: an archive hides a project and does not end its work
 * (kit-tenancy `suspension.ts`).
 */
async function projectScopeOf(
  ctx: EventContext,
  payload: { orgId: string; workspaceId: string; projectId: string },
): Promise<KnowledgeScope & { level: "project" }> {
  const ambient = ctx.container.resolve(KNOWLEDGE_SCOPE)
  if (ambient.orgId !== payload.orgId) {
    throw internal(`a knowledge event of organization ${payload.orgId} was delivered in ${ambient.orgId}`)
  }
  const data = ctx.container.resolve(DATA_SOURCE)
  const inOrg = { level: "org" as const, actor: ambient.actor, orgId: ambient.orgId }
  if (!(await data.forScope(inOrg).of(workspaces).get(payload.workspaceId))) {
    throw internal(
      `a knowledge event names workspace ${payload.workspaceId}, which organization ${ambient.orgId} does not have`,
    )
  }
  const inWorkspace = {
    level: "workspace" as const,
    actor: ambient.actor,
    orgId: ambient.orgId,
    workspaceId: payload.workspaceId,
  }
  if (!(await data.forScope(inWorkspace).of(projects).get(payload.projectId))) {
    throw internal(
      `a knowledge event names project ${payload.projectId}, which is not in workspace ${payload.workspaceId}`,
    )
  }
  return {
    level: "project",
    actor: ambient.actor,
    orgId: ambient.orgId,
    workspaceId: payload.workspaceId,
    projectId: payload.projectId,
  }
}

export const indexJob: JobDef<DocumentJobPayload> = {
  payload: documentPayload,
  async handler(payload, ctx) {
    const scope = ctx.container.resolve(KNOWLEDGE_SCOPE)
    if (scope.level !== "project") throw internal(`a knowledge job runs in a project scope, not a ${scope.level} one`)
    await indexDocument(
      {
        ...knowledgeSearchDeps(ctx.container),
        beforeIndex: async (subject) => {
          if (!ctx.signal || !ctx.container.has(NATIVE_JOB_ADMISSION)) throw serviceRefusal("observation_missing")
          ctx.signal.throwIfAborted()
          await ctx.container.resolve(NATIVE_JOB_ADMISSION).authorize(scope, INDEX_JOB, subject, ctx.signal)
        },
      },
      scope,
      payload.documentId,
      {
        ...(ctx.signal === undefined ? {} : { signal: ctx.signal }),
        ...(ctx.finalAttempt === undefined ? {} : { finalAttempt: ctx.finalAttempt }),
      },
    )
  },
}

/**
 * The automatic PDF intake (core-11): the files module enqueues it for every stored upload this module's
 * processor wants, in the upload's project. It acts for the administrator who opted the project in, so
 * the worker's native admission checks that administrator's permission and the platform's services again
 * before each step that leaves the process, as the index job does for the person who asked.
 */
export const autoIndexJob: JobDef<ProcessorJobPayload> = {
  payload: processorJobPayloadSchema,
  async handler(payload, ctx) {
    const scope = ctx.container.resolve(KNOWLEDGE_SCOPE)
    if (scope.level !== "project") throw internal(`a knowledge job runs in a project scope, not a ${scope.level} one`)
    await autoIndexUpload(
      {
        ...knowledgeSearchDeps(ctx.container),
        beforeIndex: async (subject) => {
          if (!ctx.signal || !ctx.container.has(NATIVE_JOB_ADMISSION)) throw serviceRefusal("observation_missing")
          ctx.signal.throwIfAborted()
          await ctx.container.resolve(NATIVE_JOB_ADMISSION).authorize(scope, AUTO_INDEX_JOB, subject, ctx.signal)
        },
      },
      scope,
      payload,
      {
        ...(ctx.signal === undefined ? {} : { signal: ctx.signal }),
        ...(ctx.finalAttempt === undefined ? {} : { finalAttempt: ctx.finalAttempt }),
      },
    )
  },
}

export const removeJob: JobDef<DocumentJobPayload> = {
  payload: documentPayload,
  async handler(payload, ctx) {
    const scope = ctx.container.resolve(KNOWLEDGE_SCOPE)
    if (scope.level !== "project") throw internal(`a knowledge job runs in a project scope, not a ${scope.level} one`)
    await removeDocument(knowledgeSearchDeps(ctx.container), scope, payload.documentId, payload.storageKey)
  },
}

async function onDocumentIndexRequested(
  event: {
    name: string
    payload: {
      orgId: string
      workspaceId: string
      projectId: string
      documentId: string
      collectionName: string
      fileId: string
    }
  },
  ctx: EventContext,
): Promise<void> {
  const scope = await projectScopeOf(ctx, event.payload)
  await ctx.container.resolve(JOB_QUEUE).enqueue(
    scope,
    INDEX_JOB,
    {
      documentId: event.payload.documentId,
      collectionName: event.payload.collectionName,
      fileId: event.payload.fileId,
    },
    {
      key: indexJobKey(event.payload.collectionName, event.payload.fileId),
    },
  )
}

async function onDocumentRemoveRequested(
  event: {
    name: string
    payload: {
      orgId: string
      workspaceId: string
      projectId: string
      documentId: string
      collectionName: string
      fileId: string
      storageKey?: string | undefined
    }
  },
  ctx: EventContext,
): Promise<void> {
  const scope = await projectScopeOf(ctx, event.payload)
  await ctx.container.resolve(JOB_QUEUE).enqueue(
    scope,
    REMOVE_JOB,
    {
      documentId: event.payload.documentId,
      collectionName: event.payload.collectionName,
      fileId: event.payload.fileId,
      ...(event.payload.storageKey === undefined ? {} : { storageKey: event.payload.storageKey }),
    },
    {
      key: removeJobKey(event.payload.collectionName, event.payload.fileId),
    },
  )
}

async function onSourceDeleted(
  event: { payload: { orgId: string; workspaceId: string; projectId: string; fileId: string; storageKey: string } },
  ctx: EventContext,
): Promise<void> {
  const scope = await projectScopeOf(ctx, event.payload)
  // The file's row is gone by now: the key it had travels with the removal (F407).
  await requestSourceRemoval(knowledgeDeps(ctx.container), scope, event.payload.fileId, event.payload.storageKey)
}

export const KNOWLEDGE_JOBS = { [INDEX_JOB]: indexJob, [REMOVE_JOB]: removeJob, [AUTO_INDEX_JOB]: autoIndexJob }
export const KNOWLEDGE_SUBSCRIPTIONS = [
  subscribe(fileDeleted, onSourceDeleted),
  subscribe(documentIndexRequested, onDocumentIndexRequested),
  subscribe(documentRemoveRequested, onDocumentRemoveRequested),
]