BPOS-MVP/apps/api/src/modules/ledger/internal/usecase/ledger-service.ts

/**
 * ledger 用例 —— 编排 domain,依赖 port.ts,不 import 任何 adapter(编码规范 §5.3)。
 *
 * 写三分的不变量在此落实(OS 09):
 *   systemWrite  → 正式版 + **即时**数据事件
 *   draftWrite   → 草稿层,**零**事件
 *   releaseWrite → 冲突检查 → 原子合并 → **此刻**抛事件
 */
import type { EnvelopeDraft, LedgerPage, LedgerQuery, ObjectDeclaration, TaskId } from '@contracts';
import { eventIdSchema } from '@contracts';
import type {
  DraftWrite,
  LedgerService,
  ObjectAddress,
  ReleaseWriteResult,
  SchemaApplyResult,
  SystemWrite,
} from '../../contract.ts';
import { DraftConflictError, ObjectNotMaterializedError } from '../../contract.ts';
import type {
  Clock,
  DraftStore,
  EventPublisher,
  IdGenerator,
  LedgerStore,
  ObjectRef,
  SimulationProbe,
} from '../../port.ts';
import { mergeDeclarations, planIsNonDestructive, planSchemaChanges } from '../domain/schema-merge.ts';
import { detectConflicts, keyOf, overlay, pickKey } from '../domain/task-view.ts';

export type LedgerServiceDeps = {
  readonly store: LedgerStore;
  readonly drafts: DraftStore;
  readonly events: EventPublisher;
  readonly clock: Clock;
  readonly ids: IdGenerator;
  /** 试跑判定 —— 放行写据此把主数据挡在正式表外(见 port.ts `SimulationProbe`) */
  readonly simulation: SimulationProbe;
};

function groupDeclarations(declarations: readonly ObjectDeclaration[]): Map<string, ObjectDeclaration[]> {
  const groups = new Map<string, ObjectDeclaration[]>();
  for (const d of declarations) {
    const key = `${d.domain}.${d.object}`;
    const bucket = groups.get(key);
    if (bucket === undefined) groups.set(key, [d]);
    else bucket.push(d);
  }
  return groups;
}

export function createLedgerService(deps: LedgerServiceDeps): LedgerService {
  function dataEvent(
    ref: ObjectRef,
    key: Record<string, string | number>,
    changed: readonly string[],
    operation: 'insert' | 'update' | 'delete',
    taskId: TaskId | undefined,
    cause: { kind: 'ledger-write'; ref: string },
  ): EnvelopeDraft {
    return {
      id: eventIdSchema.parse(deps.ids.next()),
      type: 'data',
      correlation: {
        // 未绑任务的系统写不指定消费目标 —— 三级消费由 `event` 判定(OS 6.3)
        ...(taskId === undefined ? {} : { target: { kind: 'task' as const, id: taskId } }),
        cause,
        keys: Object.fromEntries(Object.entries(key).map(([k, v]) => [k, String(v)])),
      },
      payload: {
        data: {
          domain: ref.domain,
          object: ref.object,
          primaryKey: key,
          changed: [...changed],
          operation,
        },
      },
      occurredAt: deps.clock.now().toISOString(),
      source: { kind: 'engine', detail: 'ledger' },
    };
  }

  async function requireSchema(address: ObjectAddress) {
    const described = await deps.store.describe(address);
    if (described === undefined) throw new ObjectNotMaterializedError(`${address.domain}.${address.object}`);
    return described;
  }

  return {
    async applyDeclarations(declarations) {
      const results: SchemaApplyResult[] = [];
      for (const [, group] of groupDeclarations(declarations)) {
        const first = group[0]!;
        const ref: ObjectRef = { domain: first.domain, object: first.object };
        const merge = mergeDeclarations(group);
        if (merge.merged === undefined || merge.conflicts.length > 0) {
          results.push({ merged: merge.merged, notices: merge.notices, conflicts: merge.conflicts, applied: false });
          continue;
        }
        const current = await deps.store.describe(ref);
        const plan = planSchemaChanges(merge.merged, current);
        const conflicts = [...merge.conflicts, ...plan.conflicts];
        if (conflicts.length > 0 || !planIsNonDestructive(plan)) {
          results.push({ merged: merge.merged, notices: [...merge.notices, ...plan.notices], conflicts, applied: false });
          continue;
        }
        if (plan.changes.length > 0) await deps.store.applyChanges(ref, plan.changes);
        /*
         * 性质每次都记 —— 它随声明走,不随建表走。
         * 声明合并已经定了口径(平台的 master 优先于活动的缺省 record),这里照写。
         */
        await deps.store.setObjectMeta(ref, merge.merged.kind, merge.merged.caseKey);
        results.push({
          merged: merge.merged,
          notices: [...merge.notices, ...plan.notices],
          conflicts: [],
          applied: true,
        });
      }
      return results;
    },

    async describe(address) {
      return deps.store.describe(address);
    },

    async listObjects() {
      return deps.store.listObjects();
    },

    /** ① 系统写:世界已发生的事实,免门,即时抛事件 */
    async systemWrite(write: SystemWrite) {
      await requireSchema(write.address);
      await deps.store.upsert(write.address, write.key, write.values);
      await deps.events.publish(
        dataEvent(
          write.address,
          write.key,
          Object.keys(write.values),
          'update',
          write.taskId,
          { kind: 'ledger-write', ref: `${write.cause.kind}:${write.cause.ref}` },
        ),
      );
    },

    /** ② 草稿写:读己之写,正式版不可见,**不产生数据事件** */
    async draftWrite(write: DraftWrite) {
      const schema = await requireSchema(write.address);
      const existing = await deps.store.selectByKeys(write.address, [write.key]);
      const baseline: Record<string, unknown> = {};
      const row = existing[0];
      if (row !== undefined) {
        for (const field of Object.keys(write.patch)) baseline[field] = row[field];
      }
      await deps.drafts.put(write.taskId, write.address, {
        key: pickKey({ ...write.key }, schema.primaryKey.length > 0 ? schema.primaryKey : Object.keys(write.key)),
        patch: write.patch,
        tombstone: write.tombstone ?? false,
        baseline,
      });
    },

    /**
     * 撤掉本单草稿层的一行 —— **只撤自己写过的**。
     *
     * 回 `false` 表示草稿层里根本没有这一行:那多半是主键写法对不上
     * (`DAG…` 与 `DAG…-R0` 是两行),如实说出来比静默成功有用 ——
     * 静默成功会让模型以为撤掉了,下一拍判据照旧不过,它又去撤一次。
     */
    async draftDrop(drop) {
      const mine = await deps.drafts.listForObject(drop.taskId, drop.address);
      const schema = await requireSchema(drop.address);
      const keys = schema.primaryKey.length > 0 ? schema.primaryKey : Object.keys(drop.key);
      const wanted = pickKey({ ...drop.key }, keys);
      const hit = mine.some((d) => keys.every((k) => String((d.key as Record<string, unknown>)[k] ?? '') === String(wanted[k] ?? '')));
      if (!hit) return false;
      await deps.drafts.drop(drop.taskId, drop.address, wanted);
      return true;
    },

    /** ③ 放行写:冲突检查 → 原子合并 → 此刻抛事件 */
    async releaseWrite(taskId): Promise<ReleaseWriteResult> {
      const conflicts = await this.checkConflicts(taskId);
      if (conflicts.length > 0) throw new DraftConflictError(conflicts);

      const drafts = await deps.drafts.list(taskId);
      if (drafts.length === 0) return { mergedRows: 0, emittedEvents: 0, simulatedRows: 0, merged: [] };

      /*
       * **试跑单不落主数据** —— 挡在这一处,是因为这里是草稿层通往世界的唯一一道口子。
       *
       * 试跑照旧是真跑:过装载两审、过预算、过放行门,单据照写。挡的只有主数据 ——
       * 它按业务实体分行,试跑往 `customer.profile` 建一位客户,那位客户就永久留在库里,
       * 「新客户来函」那条样例第二次跑就成了老客户,这条路从此试不出来。
       * 单据不挡:它按单标识分行,一次试跑写自己那一行,清运行数据就干净了。
       *
       * 被挡下的行仍然进 `merged`(出口载荷取的就是它)—— 拦的是**落到世界**,
       * 不是拦这一单自己看得见:活动缓存里它一直在,这一单该怎么跑还怎么跑。
       */
      const simulated = await deps.simulation.isSimulated(taskId);
      const masters = new Set<string>();
      if (simulated) {
        for (const ref of await deps.store.masterRefs()) masters.add(`${ref.domain}.${ref.object}`);
      }
      const heldBack = (draft: { domain: string; object: string }): boolean =>
        simulated && masters.has(`${draft.domain}.${draft.object}`);
      let simulatedRows = 0;

      const pending: EnvelopeDraft[] = [];
      /*
       * 合并后**读回来**的行 —— 「落地即事实」,事实是表里那一份,不是写进去之前的草稿。
       *
       * 从前这里直接回 `draft.patch`,于是出口载荷可以跟台账不一致:草稿里是字符串
       * `"true"`(人的回传值天生是字符串),列是 boolean,PG 存的时候转了,可发出去的口里
       * 还是字符串 —— 下游活动的入参校验当场拒收「/needSourcing must be boolean」,
       * 节点起不来、报 blocked,而台账里明明是对的(实测)。
       * 读回来之后,出口载荷与台账逐字一致,类型也由列定义说了算。
       */
      const landed: Array<{ domain: string; object: string; key: Record<string, unknown>; values: Record<string, unknown> }> = [];

      await deps.store.transaction(async (store, draftStore) => {
        for (const draft of drafts) {
          const ref: ObjectRef = { domain: draft.domain, object: draft.object };
          if (heldBack(draft)) {
            simulatedRows += 1;
            if (draft.tombstone) continue; // 试跑不删主数据,连一行也不许少
            /*
             * 值取「表里现有的 + 这一单改的」——没真写进去,读不回 PG 归型后的那一份,
             * 所以这里的值是草稿的原样(人的回传天生是字符串)。如实记在这:
             * 主数据不进出口载荷,下游拿不到它,归不归型都影响不到谁。
             */
            const current = (await store.selectByKeys(ref, [draft.key]))[0];
            landed.push({
              domain: draft.domain,
              object: draft.object,
              key: draft.key,
              values: { ...(current ?? {}), ...draft.patch },
            });
            continue;
          }
          if (draft.tombstone) {
            await store.deleteRow(ref, draft.key);
            pending.push(
              dataEvent(ref, draft.key, [], 'delete', taskId, { kind: 'ledger-write', ref: `release:${taskId}` }),
            );
            continue;
          }
          const existing = await store.selectByKeys(ref, [draft.key]);
          await store.upsert(ref, draft.key, draft.patch);
          const stored = (await store.selectByKeys(ref, [draft.key]))[0];
          landed.push({
            domain: draft.domain,
            object: draft.object,
            key: draft.key,
            // 读不回来只能退回草稿(理论上不该发生:刚 upsert 过);宁可退也不丢这一行
            values: stored ?? draft.patch,
          });
          pending.push(
            dataEvent(
              ref,
              draft.key,
              Object.keys(draft.patch),
              existing.length === 0 ? 'insert' : 'update',
              taskId,
              { kind: 'ledger-write', ref: `release:${taskId}` },
            ),
          );
        }
        await draftStore.clear(taskId);
      });

      for (const envelope of pending) await deps.events.publish(envelope);
      return {
        mergedRows: drafts.length - simulatedRows,
        emittedEvents: pending.length,
        simulatedRows,
        merged: landed,
      };
    },

    async query(query: LedgerQuery): Promise<LedgerPage> {
      const { rows, total } = await deps.store.select(query);
      return { rows, total };
    },

    async taskView(taskId, query): Promise<LedgerPage> {
      const ref: ObjectRef = { domain: query.address.domain, object: query.address.object };
      const schema = await requireSchema(ref);
      const [official, drafts] = await Promise.all([
        deps.store.select(query),
        deps.drafts.listForObject(taskId, ref),
      ]);
      const rows = overlay(official.rows, drafts, schema.primaryKey);
      return { rows, total: rows.length };
    },

    async taskDrafts(taskId) {
      const drafts = await deps.drafts.list(taskId);
      return drafts.map((d) => ({ domain: d.domain, object: d.object, key: d.key, patch: d.patch }));
    },

    async checkConflicts(taskId) {
      const drafts = await deps.drafts.list(taskId);
      if (drafts.length === 0) return [];
      const byObject = new Map<string, Array<(typeof drafts)[number]>>();
      for (const d of drafts) {
        const k = `${d.domain}.${d.object}`;
        const bucket = byObject.get(k);
        if (bucket === undefined) byObject.set(k, [d]);
        else bucket.push(d);
      }
      const findings = [];
      for (const [, group] of byObject) {
        const first = group[0]!;
        const ref: ObjectRef = { domain: first.domain, object: first.object };
        const schema = await requireSchema(ref);
        const rows = await deps.store.selectByKeys(
          ref,
          group.map((g) => g.key),
        );
        const officialByKey = new Map(rows.map((r) => [keyOf(pickKey(r, schema.primaryKey)), r]));
        findings.push(...detectConflicts(group, officialByKey));
      }
      return findings;
    },

    async discardDrafts(taskId) {
      await deps.drafts.clear(taskId);
    },
  };
}