BPOS-TMP-REPO/apps/api/src/app/__tests__/activity-ops-store.it.test.ts

/**
 * 启动幂等键在真库上的并发:两条连接同时占同一枚键,**恰一个赢**;单没建成的过期键可被重占。
 * 独占一个库(照 mail-store.it 的先例):建表 / 清表都在 `jcai_it_activity_ops` 里,谁的库都不碰。
 */
import { readFileSync } from 'node:fs';
import { afterAll, beforeAll, describe, expect, it } from 'vitest';
import { Pool } from 'pg';
import { drizzle } from 'drizzle-orm/node-postgres';
import { sql } from 'drizzle-orm';
import type { Db } from '@shared/db.ts';
import { createActivityOpsStore } from '../activity-ops/store.ts';

const IT_DB = 'jcai_it_activity_ops';

async function provisionDatabase(): Promise<{ url: string; stop(): Promise<void> } | undefined> {
  const base = process.env['JCAI_TEST_DATABASE_URL'] ?? process.env['DATABASE_URL'] ?? 'postgres://jcai:jcai@localhost:54329/jcai';
  try {
    const admin = new Pool({ connectionString: base, max: 1, connectionTimeoutMillis: 1500 });
    const exists = await admin.query(`select 1 from pg_database where datname = $1`, [IT_DB]);
    if (exists.rowCount === 0) await admin.query(`create database ${IT_DB}`);
    await admin.end();
    return { url: base.replace(/\/[^/]*$/, `/${IT_DB}`), stop: async () => undefined };
  } catch {
    /* 开发实例连不上,试 Testcontainers */
  }
  try {
    const { PostgreSqlContainer } = await import('@testcontainers/postgresql');
    const started = await new PostgreSqlContainer('postgres:16-alpine').start();
    return { url: started.getConnectionUri(), stop: () => started.stop().then(() => undefined) };
  } catch (error) {
    process.stderr.write(`⚠️  activity-ops store it 跳过:既连不上开发实例也起不了 Testcontainers(${error instanceof Error ? error.message : String(error)})\n`);
    return undefined;
  }
}

const database = await provisionDatabase();
let poolA: Pool | undefined;
let poolB: Pool | undefined;
let dbA: Db;
let dbB: Db;

const MIGRATION = new URL('../../../../../db/migrations/0035_activity_ops.sql', import.meta.url);

beforeAll(async () => {
  if (database === undefined) return;
  poolA = new Pool({ connectionString: database.url, max: 2 });
  poolB = new Pool({ connectionString: database.url, max: 2 });
  dbA = drizzle(poolA);
  dbB = drizzle(poolB);
  await poolA.query('create schema if not exists sys');
  await poolA.query('drop table if exists sys.activity_launch; drop table if exists sys.activity_profile');
  await poolA.query(readFileSync(MIGRATION, 'utf8'));
});

afterAll(async () => {
  await poolA?.end();
  await poolB?.end();
  await database?.stop();
});

describe.skipIf(database === undefined)('activity_launch 幂等键(真库)', () => {
  it('两条连接同时占同一枚键,恰一个 claimed,另一个读回那一行;绑定后重放拿到 task_id', async () => {
    const a = createActivityOpsStore(dbA);
    const b = createActivityOpsStore(dbB);
    const input = { key: 'race-1', activityId: 'w05', ownerId: 'owner', launchedBy: 'me' };
    const results = await Promise.all(Array.from({ length: 20 }, (_, i) => (i % 2 === 0 ? a : b).claimLaunch(input)));
    expect(results.filter((r) => r === 'claimed')).toHaveLength(1);
    for (const r of results) if (r !== 'claimed') expect(r).toMatchObject({ key: 'race-1', activityId: 'w05', taskId: null });
    await a.bindLaunch('race-1', 't-1');
    expect(await b.claimLaunch(input)).toMatchObject({ taskId: 't-1' });
    expect(await b.launchOf('t-1')).toMatchObject({ key: 'race-1', launchedBy: 'me' });
  });

  it('单没建成的键:没过期不许重占;过期后恰一个重占成功;已绑 task 的永远不许', async () => {
    const a = createActivityOpsStore(dbA);
    const b = createActivityOpsStore(dbB);
    expect(await a.claimLaunch({ key: 'stale-1', activityId: 'w05', ownerId: 'o', launchedBy: 'x' })).toBe('claimed');
    expect(await b.reclaimLaunch('stale-1', 60_000, 'o', 'y')).toBe(false);
    await dbA.execute(sql`update sys.activity_launch set claimed_at = now() - interval '2 minutes' where key = 'stale-1'`);
    const won = await Promise.all([a.reclaimLaunch('stale-1', 60_000, 'o', 'y'), b.reclaimLaunch('stale-1', 60_000, 'o', 'z')]);
    expect(won.filter(Boolean)).toHaveLength(1);
    await a.bindLaunch('stale-1', 't-2');
    await dbA.execute(sql`update sys.activity_launch set claimed_at = now() - interval '2 minutes' where key = 'stale-1'`);
    expect(await b.reclaimLaunch('stale-1', 60_000, 'o', 'q')).toBe(false);
  });

  it('活动档案:upsert 冲突即更新,负责人可空', async () => {
    const a = createActivityOpsStore(dbA);
    await a.upsertProfile({ activityId: 'w05', ownerId: null, description: '一', ico: 'git-compare', hue: 'teal' });
    await a.upsertProfile({ activityId: 'w05', ownerId: 'owner', description: '二', ico: 'git-compare', hue: 'teal' });
    expect((await a.profilesOf()).get('w05')).toEqual({ activityId: 'w05', ownerId: 'owner', description: '二', ico: 'git-compare', hue: 'teal' });
  });
});