import assert from "node:assert/strict";
import { spawnSync } from "node:child_process";
import { fileURLToPath } from "node:url";
import test from "node:test";
import { Client } from "pg";

import { startPostgresFixture } from "./helpers/postgres-fixture.ts";

const runnerPath = fileURLToPath(new URL("../scripts/db-migrate.mjs", import.meta.url));
const ids = {
  owner: "10000000-0000-4000-8000-000000000001",
  billing: "10000000-0000-4000-8000-000000000002",
  model: "10000000-0000-4000-8000-000000000003",
  support: "10000000-0000-4000-8000-000000000004",
  auditor: "10000000-0000-4000-8000-000000000005",
  subscriber: "20000000-0000-4000-8000-000000000001",
  trialUser: "20000000-0000-4000-8000-000000000002",
  annualUser: "20000000-0000-4000-8000-000000000003",
  creditUser: "20000000-0000-4000-8000-000000000004",
  snapshotUser: "20000000-0000-4000-8000-000000000005",
  failedGrantUser: "20000000-0000-4000-8000-000000000006",
  concurrentRenewalUser: "20000000-0000-4000-8000-000000000007",
  concurrentOverlapUser: "20000000-0000-4000-8000-000000000008",
};

function orderNo(suffix: string) {
  return `JYTEST${suffix.padEnd(18, "0")}`;
}

function sqlJson(value: unknown) {
  return JSON.stringify(value).replaceAll("'", "''");
}

test("billing, subscriptions, usage authorization, RBAC, and model publication remain transactional", async () => {
  const fixture = startPostgresFixture();
  const sql = (statement: string) => fixture.psql(statement);
  const sqlAsOwner = (statement: string) => fixture.psqlAs(
    "schema_owner",
    "schema-owner-test-password",
    statement,
  );
  const expectSqlError = (statement: string, pattern: RegExp) => {
    assert.throws(() => sqlAsOwner(statement), pattern);
  };
  const expectAdminRuntimeError = (statement: string, pattern: RegExp) => {
    assert.throws(
      () => fixture.psqlAs("admin_runtime", "admin-runtime-test-password", statement),
      pattern,
    );
  };
  const insertOrder = (input: {
    suffix: string;
    userId: string;
    productId: string;
    code: string;
    version?: number;
    money: number;
    credits?: number;
  }) => sql(`
    insert into public.payment_orders (
      order_no,user_id,package_id,product_id,product_code,product_version,
      product_snapshot,entitlement_snapshot,money_cents,currency,credits,grant_status
    )
    select
      '${orderNo(input.suffix)}','${input.userId}',null,p.id,p.code,p.version,
      jsonb_build_object(
        'id',p.id,'code',p.code,'version',p.version,'name',p.name,'description',p.description,
        'productType',p.product_type,'billingPeriod',p.billing_period,'intervalCount',p.interval_count,
        'priceCents',p.price_cents,'currency',p.currency,'oneTimePerUser',p.one_time_per_user
      ),
      coalesce((select jsonb_agg(jsonb_build_object(
        'featureKey',e.feature_key,'allowanceType',e.allowance_type,'allowanceCount',e.allowance_count,
        'resetPeriod',e.reset_period,'modelTier',e.model_tier,'fairUsePolicyId',e.fair_use_policy_id,
        'metadata',e.metadata
      ) order by e.feature_key) from public.product_entitlements e where e.product_id=p.id),'[]'::jsonb),
      ${input.money},'CNY',${input.credits ?? 0},'pending'
    from public.billing_products p
    where p.id='${input.productId}' and p.code='${input.code}' and p.version=${input.version ?? 1}
    returning id
  `);
  const settleConcurrently = async (orders: Array<{ suffix: string; trade: string; money: number; hash: string }>) => {
    const clients = orders.map(() => new Client({
      connectionString: fixture.connectionUrl("postgres", "postgres-test-password"),
    }));
    await Promise.all(clients.map((client) => client.connect()));
    try {
      return await Promise.all(orders.map((order, index) => clients[index].query(
        "select * from public.settle_order($1,$2,$3,$4)",
        [orderNo(order.suffix), order.trade, order.money, order.hash],
      )));
    } finally {
      await Promise.all(clients.map((client) => client.end()));
    }
  };
  const completeUsageConcurrently = async (userId: string, requestId: string, events: Array<Record<string, unknown>>) => {
    const clients = events.map(() => new Client({
      connectionString: fixture.connectionUrl("postgres", "postgres-test-password"),
    }));
    await Promise.all(clients.map((client) => client.connect()));
    try {
      return await Promise.all(events.map((event, index) => clients[index].query(
        "select * from public.complete_usage($1,$2,$3::jsonb)",
        [userId, requestId, JSON.stringify(event)],
      )));
    } finally {
      await Promise.all(clients.map((client) => client.end()));
    }
  };

  try {
    const migration = spawnSync(process.execPath, [runnerPath], {
      encoding: "utf8",
      env: {
        ...process.env,
        SCHEMA_DATABASE_URL: fixture.connectionUrl("schema_owner", "schema-owner-test-password"),
      },
    });
    assert.equal(migration.status, 0, migration.stderr);
    assert.match(migration.stdout, /applied 20260807160000_model_provider_encrypted_credentials\.sql/);

    fixture.psqlAs("identity_runtime", "identity-runtime-test-password", `
      insert into identity.users (id,name,email,email_verified,email_verified_at,role) values
        ('${ids.owner}','Owner','owner@example.com',true,now(),'admin'),
        ('${ids.billing}','Billing','billing@example.com',true,now(),'user'),
        ('${ids.model}','Model','model@example.com',true,now(),'user'),
        ('${ids.support}','Support','support@example.com',true,now(),'user'),
        ('${ids.auditor}','Auditor','auditor@example.com',true,now(),'user'),
        ('${ids.subscriber}','Subscriber','subscriber@example.com',true,now(),'user'),
        ('${ids.trialUser}','Trial','trial@example.com',true,now(),'user'),
        ('${ids.annualUser}','Annual','annual@example.com',true,now(),'user'),
        ('${ids.creditUser}','Credit','credit@example.com',true,now(),'user'),
        ('${ids.snapshotUser}','Snapshot','snapshot@example.com',true,now(),'user'),
        ('${ids.failedGrantUser}','Failed Grant','failed-grant@example.com',true,now(),'user'),
        ('${ids.concurrentRenewalUser}','Concurrent Renewal','renewal@example.com',true,now(),'user'),
        ('${ids.concurrentOverlapUser}','Concurrent Overlap','overlap@example.com',true,now(),'user')
    `);
    sql(`
      update public.profiles set credits=10 where id in ('${ids.subscriber}','${ids.creditUser}');
      insert into public.admin_users(user_id,created_by) values
        ('${ids.owner}','${ids.owner}'),('${ids.billing}','${ids.owner}'),
        ('${ids.model}','${ids.owner}'),('${ids.support}','${ids.owner}'),('${ids.auditor}','${ids.owner}');
      insert into public.admin_user_roles(admin_user_id,role_id,assigned_by)
      select v.user_id,r.id,'${ids.owner}'::uuid
      from (values
        ('${ids.owner}'::uuid,'owner'),('${ids.billing}'::uuid,'billing_admin'),
        ('${ids.model}'::uuid,'model_admin'),('${ids.support}'::uuid,'support'),
        ('${ids.auditor}'::uuid,'auditor')
      ) v(user_id,role_code) join public.admin_roles r on r.code=v.role_code
    `);

    assert.equal(sql(`select public.admin_has_permission('${ids.billing}','billing.products.publish')`), "t");
    assert.equal(sql(`select public.admin_has_permission('${ids.billing}','models.publish')`), "f");
    assert.equal(sql(`select public.admin_has_permission('${ids.model}','models.rollback')`), "t");
    assert.equal(sql(`select public.admin_has_permission('${ids.model}','billing.adjustments.write')`), "f");
    assert.equal(sql(`select public.admin_has_permission('${ids.support}','billing.orders.read')`), "t");
    assert.equal(sql(`select public.admin_has_permission('${ids.support}','billing.products.write')`), "f");
    assert.equal(sql(`select public.admin_has_permission('${ids.auditor}','audit.read')`), "t");
    assert.equal(sql(`select public.admin_has_permission('${ids.auditor}','ops.flags.write')`), "f");
    assert.equal(sql(`select public.admin_has_permission('${ids.owner}','admin.customers.birth_data.read')`), "t");
    assert.equal(sql(`select public.admin_has_permission('${ids.support}','admin.customers.read')`), "t");
    assert.equal(sql(`select public.admin_has_permission('${ids.support}','admin.customers.birth_data.read')`), "f");
    assert.equal(sql("select pg_has_role('admin_runtime','service_role','MEMBER')"), "f");
    assert.equal(sql("select has_function_privilege('admin_runtime','public.admin_permission_keys(uuid)','execute')"), "t");
    assert.equal(sql("select has_column_privilege('admin_runtime','public.profiles','birth_date','select')"), "f");
    expectAdminRuntimeError("set role service_role", /permission denied to set role/);
    expectSqlError(
      `select * from public.admin_manage_role('${ids.owner}','${ids.owner}','owner',false,'不得移除最后 Owner','last-owner')`,
      /last_owner_protected/,
    );
    expectSqlError(
      `update public.admin_users set revoked_at=clock_timestamp(),revoked_by='${ids.owner}' where user_id='${ids.owner}'`,
      /last_owner_protected/,
    );
    sql(`select * from public.admin_manage_role('${ids.owner}','${ids.billing}','owner',true,'增加第二位 Owner','second-owner')`);
    sql(`update public.admin_users set revoked_at=clock_timestamp(),revoked_by='${ids.billing}' where user_id='${ids.owner}'`);
    assert.equal(sql(`select public.admin_has_permission('${ids.owner}','admin.access')`), "f");
    expectSqlError(
      `update public.admin_users set revoked_at=clock_timestamp(),revoked_by='${ids.billing}' where user_id='${ids.billing}'`,
      /last_owner_protected/,
    );
    expectSqlError(
      `delete from public.admin_user_roles where admin_user_id='${ids.billing}' and role_id=(select id from public.admin_roles where code='owner')`,
      /last_owner_protected/,
    );
    sql(`update public.admin_users set revoked_at=null,revoked_by=null where user_id='${ids.owner}'`);
    sql(`delete from public.admin_user_roles where admin_user_id='${ids.billing}' and role_id=(select id from public.admin_roles where code='owner')`);

    sql(`
      update public.profiles
      set birth_date='1990-01-02', birth_time_status='reported', birth_place_label='Taipei'
      where id='${ids.subscriber}'
    `);
    expectAdminRuntimeError(
      `select * from public.admin_read_customer_birth_data('${ids.support}',array['${ids.subscriber}'::uuid],'support-birth-read')`,
      /admin_permission_denied/,
    );
    assert.equal(
      fixture.psqlAs(
        "admin_runtime",
        "admin-runtime-test-password",
        `select birth_date||':'||birth_time_status||':'||birth_place_label
         from public.admin_read_customer_birth_data('${ids.owner}',array['${ids.subscriber}'::uuid],'owner-birth-read')`,
      ),
      "1990-01-02:reported:Taipei",
    );
    assert.equal(
      sql(`select action||':'||permission_used from audit.admin_audit_logs where request_id='owner-birth-read'`),
      "admin.customer.birth_data.read:admin.customers.birth_data.read",
    );

    sql(`
      insert into public.admin_role_permissions(role_id,permission_id)
      select r.id,p.id from public.admin_roles r cross join public.admin_permissions p
      where r.code='support' and p.permission_key='admin.users.manage_roles'
      on conflict do nothing;
      insert into public.admin_user_roles(admin_user_id,role_id,assigned_by)
      select '${ids.billing}',id,'${ids.owner}' from public.admin_roles where code='owner'
      on conflict do nothing
    `);
    const ownerRaceClients = Array.from({ length: 3 }, () => new Client({
      connectionString: fixture.connectionUrl("postgres", "postgres-test-password"),
    }));
    await Promise.all(ownerRaceClients.map(async (client) => {
      await client.connect();
      await client.query("set plpgsql.variable_conflict = use_column");
    }));
    const [blocker, firstOwnerRevoke, secondOwnerRevoke] = ownerRaceClients;
    let revocations: Promise<unknown>[] = [];
    try {
      await blocker.query("begin");
      await blocker.query("select pg_advisory_xact_lock(1096040772, 1)");
      const first = firstOwnerRevoke.query(
        "select * from public.admin_manage_role($1,$2,'owner',false,$3,$4)",
        [ids.support, ids.owner, "并发移除 Owner A", "owner-race-a"],
      );
      const second = secondOwnerRevoke.query(
        "select * from public.admin_manage_role($1,$2,'owner',false,$3,$4)",
        [ids.support, ids.billing, "并发移除 Owner B", "owner-race-b"],
      );
      revocations = [first, second];
      let waiters = 0;
      for (let attempt = 0; attempt < 40 && waiters < 2; attempt += 1) {
        await new Promise((resolve) => setTimeout(resolve, 25));
        const result = await blocker.query<{ count: string }>(`
          select count(*)::text as count from pg_stat_activity
          where pid <> pg_backend_pid()
            and query like '%admin_manage_role%'
            and wait_event_type='Lock'
            and wait_event='advisory'
        `);
        waiters = Number(result.rows[0]?.count ?? 0);
      }
      assert.equal(waiters, 2, "both Owner revocations must wait on the fixed advisory lock");
      await blocker.query("commit");
      const outcomes = await Promise.allSettled([first, second]);
      assert.equal(
        outcomes.filter((outcome) => outcome.status === "fulfilled").length,
        1,
        outcomes.map((outcome) => outcome.status === "fulfilled" ? "fulfilled" : String(outcome.reason?.message)).join(" | "),
      );
      const rejected = outcomes.find((outcome): outcome is PromiseRejectedResult => outcome.status === "rejected");
      assert.match(String(rejected?.reason?.message), /last_owner_protected/);
      assert.equal(sql(`select count(*) from public.admin_user_roles ur join public.admin_roles r on r.id=ur.role_id where r.code='owner'`), "1");
    } finally {
      await blocker.query("rollback").catch(() => undefined);
      await Promise.allSettled(revocations);
      await Promise.all(ownerRaceClients.map((client) => client.end()));
    }
    sql(`
      insert into public.admin_user_roles(admin_user_id,role_id,assigned_by)
      select '${ids.owner}',id,'${ids.owner}' from public.admin_roles where code='owner'
      on conflict do nothing;
      delete from public.admin_user_roles
      where admin_user_id='${ids.billing}' and role_id=(select id from public.admin_roles where code='owner');
      delete from public.admin_role_permissions
      where role_id=(select id from public.admin_roles where code='support')
        and permission_id=(select id from public.admin_permissions where permission_key='admin.users.manage_roles')
    `);

    const creditProduct = "30000000-0000-4000-8000-000000000001";
    sql(`
      insert into public.billing_products(id,code,version,name,product_type,billing_period,interval_count,price_cents,enabled,status,effective_from)
      values('${creditProduct}','credits_10',1,'10 点积分','credit_pack','none',0,1000,true,'published',now());
      insert into public.product_entitlements(product_id,feature_key,allowance_type,allowance_count,reset_period)
      values('${creditProduct}','chat.standard','credits',10,'none')
    `);
    insertOrder({ suffix: "CREDIT", userId: ids.creditUser, productId: creditProduct, code: "credits_10", money: 1000, credits: 10 });
    for (let attempt = 0; attempt < 5; attempt += 1) {
      assert.equal(sql(`select success||':'||status from public.settle_order('${orderNo("CREDIT")}','trade-credit',1000,'hash-credit')`), "true:paid");
    }
    assert.equal(sql(`select credits from public.profiles where id='${ids.creditUser}'`), "20");
    assert.equal(sql(`select count(*) from public.credit_transactions where user_id='${ids.creditUser}' and transaction_type='payment'`), "1");

    const snapshotProduct = "30000000-0000-4000-8000-000000000002";
    sql(`
      insert into public.billing_products(id,code,version,name,product_type,billing_period,interval_count,price_cents,enabled,status,effective_from)
      values('${snapshotProduct}','snapshot_monthly',1,'快照月卡','subscription','month',1,1200,true,'published',now());
      insert into public.product_entitlements(product_id,feature_key,allowance_type,allowance_count,reset_period,model_tier)
      values('${snapshotProduct}','chat.standard','quota',2,'billing_period','standard')
    `);
    insertOrder({ suffix: "SNAPSHOT", userId: ids.snapshotUser, productId: snapshotProduct, code: "snapshot_monthly", money: 1200 });
    sql(`
      update public.billing_products set billing_period='year',interval_count=2 where id='${snapshotProduct}';
      update public.product_entitlements set allowance_count=99 where product_id='${snapshotProduct}'
    `);
    assert.equal(sql(`select success||':'||status from public.settle_order('${orderNo("SNAPSHOT")}','trade-snapshot',1200,'hash-snapshot')`), "true:paid");
    assert.equal(sql(`select (ends_at=starts_at+interval '1 month')||':'||(entitlement_snapshot#>>'{0,allowanceCount}') from public.user_subscriptions where source_order_id=(select id from public.payment_orders where order_no='${orderNo("SNAPSHOT")}')`), "true:2");
    assert.equal(sql(`select (s.product_snapshot=o.product_snapshot)||':'||(s.entitlement_snapshot=o.entitlement_snapshot) from public.user_subscriptions s join public.payment_orders o on o.id=s.source_order_id where o.order_no='${orderNo("SNAPSHOT")}'`), "true:true");

    sql(`
      create function public.test_fail_subscription_grant() returns trigger language plpgsql as $$
      begin
        if new.user_id='${ids.failedGrantUser}' then raise exception 'forced_grant_failure'; end if;
        return new;
      end $$;
      create trigger test_fail_subscription_grant before insert on public.user_subscriptions
      for each row execute function public.test_fail_subscription_grant()
    `);
    insertOrder({ suffix: "FAILGRANT", userId: ids.failedGrantUser, productId: "00000000-0000-4000-8000-000000000902", code: "standard_monthly", money: 9900 });
    assert.equal(sql(`select success||':'||status from public.settle_order('${orderNo("FAILGRANT")}','trade-failed',9900,'hash-failed')`), "f:grant_failed");
    const failedPaidAt = sql(`select paid_at from public.payment_orders where order_no='${orderNo("FAILGRANT")}'`);
    assert.equal(sql(`select status||':'||grant_status||':'||(grant_error like '%forced_grant_failure%')||':'||epay_trade_no||':'||raw_notify_payload_hash||':'||(paid_at is not null) from public.payment_orders where order_no='${orderNo("FAILGRANT")}'`), "grant_pending:failed:true:trade-failed:hash-failed:true");
    assert.equal(sql(`select success||':'||status from public.settle_order('${orderNo("FAILGRANT")}','trade-failed',9900,'hash-failed-retry')`), "f:grant_failed");
    assert.equal(sql(`select count(*) from public.user_subscriptions where source_order_id=(select id from public.payment_orders where order_no='${orderNo("FAILGRANT")}')`), "0");
    sql(`drop trigger test_fail_subscription_grant on public.user_subscriptions; drop function public.test_fail_subscription_grant()`);
    assert.equal(sql(`select success||':'||status from public.settle_order('${orderNo("FAILGRANT")}','trade-failed',9900,'hash-failed-retry')`), "true:paid");
    assert.equal(sql(`select (paid_at='${failedPaidAt}'::timestamptz)||':'||raw_notify_payload_hash||':'||grant_status from public.payment_orders where order_no='${orderNo("FAILGRANT")}'`), "true:hash-failed:granted");
    assert.equal(sql(`select success||':'||status from public.settle_order('${orderNo("FAILGRANT")}','trade-failed',9900,'hash-after-grant')`), "true:paid");
    assert.equal(sql(`select raw_notify_payload_hash from public.payment_orders where order_no='${orderNo("FAILGRANT")}'`), "hash-failed");

    insertOrder({ suffix: "TRIAL1", userId: ids.trialUser, productId: "00000000-0000-4000-8000-000000000901", code: "trial_7d", money: 990 });
    assert.equal(sql(`select success||':'||status from public.settle_order('${orderNo("TRIAL1")}','trade-trial-1',990,'hash-trial-1')`), "true:paid");
    insertOrder({ suffix: "TRIAL2", userId: ids.trialUser, productId: "00000000-0000-4000-8000-000000000901", code: "trial_7d", money: 990 });
    assert.equal(sql(`select success||':'||status from public.settle_order('${orderNo("TRIAL2")}','trade-trial-2',990,'hash-trial-2')`), "f:one_time_limit");
    assert.equal(sql(`select count(*) from public.user_product_redemptions where user_id='${ids.trialUser}' and product_code='trial_7d'`), "1");

    insertOrder({ suffix: "MONTH1", userId: ids.subscriber, productId: "00000000-0000-4000-8000-000000000902", code: "standard_monthly", money: 9900 });
    assert.equal(sql(`select success||':'||status from public.settle_order('${orderNo("MONTH1")}','trade-month-1',9900,'hash-month-1')`), "true:paid");
    const monthlySubscription = sql(`select grant_reference_id from public.payment_orders where order_no='${orderNo("MONTH1")}'`);
    sql(`update public.user_subscriptions set starts_at='2026-07-31 10:00:00+00',ends_at='2026-08-31 10:00:00+00' where id='${monthlySubscription}'`);
    insertOrder({ suffix: "MONTH2", userId: ids.subscriber, productId: "00000000-0000-4000-8000-000000000902", code: "standard_monthly", money: 9900 });
    assert.equal(sql(`select success||':'||status from public.settle_order('${orderNo("MONTH2")}','trade-month-2',9900,'hash-month-2')`), "true:paid");
    assert.equal(sql(`select to_char(starts_at at time zone 'UTC','YYYY-MM-DD HH24:MI')||':'||to_char(ends_at at time zone 'UTC','YYYY-MM-DD HH24:MI') from public.user_subscriptions where source_order_id=(select id from public.payment_orders where order_no='${orderNo("MONTH2")}')`), "2026-08-31 10:00:2026-09-30 10:00");

    insertOrder({ suffix: "OVERLAP", userId: ids.subscriber, productId: "00000000-0000-4000-8000-000000000903", code: "standard_yearly", money: 59900 });
    assert.equal(sql(`select success||':'||status from public.settle_order('${orderNo("OVERLAP")}','trade-overlap',59900,'hash-overlap')`), "f:overlapping_subscription");

    insertOrder({ suffix: "YEAR1", userId: ids.annualUser, productId: "00000000-0000-4000-8000-000000000903", code: "standard_yearly", money: 59900 });
    assert.equal(sql(`select success||':'||status from public.settle_order('${orderNo("YEAR1")}','trade-year-1',59900,'hash-year-1')`), "true:paid");
    const annualSubscription = sql(`select grant_reference_id from public.payment_orders where order_no='${orderNo("YEAR1")}'`);
    sql(`update public.user_subscriptions set starts_at='2027-02-28 00:00:00+00',ends_at='2028-02-29 00:00:00+00' where id='${annualSubscription}'`);
    insertOrder({ suffix: "YEAR2", userId: ids.annualUser, productId: "00000000-0000-4000-8000-000000000903", code: "standard_yearly", money: 59900 });
    assert.equal(sql(`select success||':'||status from public.settle_order('${orderNo("YEAR2")}','trade-year-2',59900,'hash-year-2')`), "true:paid");
    assert.equal(sql(`select to_char(ends_at at time zone 'UTC','YYYY-MM-DD') from public.user_subscriptions where source_order_id=(select id from public.payment_orders where order_no='${orderNo("YEAR2")}')`), "2029-02-28");
    sql(`update public.user_subscriptions set starts_at=clock_timestamp()-interval '1 day',ends_at=clock_timestamp()+interval '1 year' where id='${annualSubscription}'`);
    assert.equal(sql(`select source||':'||credits from public.authorize_usage('${ids.annualUser}','rectification',null,'annual-monthly-release-1',1)`), "subscription:0");
    assert.equal(sql(`select success||':'||reason||':'||credits from public.authorize_usage('${ids.annualUser}','rectification',null,'annual-monthly-release-2',1)`), "f:feature_quota_exhausted:0");
    sql(`update public.user_subscriptions set starts_at=clock_timestamp()-interval '1 month 1 day' where id='${annualSubscription}'`);
    assert.equal(sql(`select source||':'||credits from public.authorize_usage('${ids.annualUser}','rectification',null,'annual-monthly-release-2',1)`), "subscription:0");

    sql(`
      create function public.test_delay_concurrent_subscription_grant() returns trigger language plpgsql as $$
      begin
        if new.user_id in ('${ids.concurrentRenewalUser}','${ids.concurrentOverlapUser}') then perform pg_sleep(0.35); end if;
        return new;
      end $$;
      create trigger test_delay_concurrent_subscription_grant before insert on public.user_subscriptions
      for each row execute function public.test_delay_concurrent_subscription_grant()
    `);
    insertOrder({ suffix: "RENEW1", userId: ids.concurrentRenewalUser, productId: "00000000-0000-4000-8000-000000000902", code: "standard_monthly", money: 9900 });
    insertOrder({ suffix: "RENEW2", userId: ids.concurrentRenewalUser, productId: "00000000-0000-4000-8000-000000000902", code: "standard_monthly", money: 9900 });
    const renewalResults = await settleConcurrently([
      { suffix: "RENEW1", trade: "trade-renew-1", money: 9900, hash: "hash-renew-1" },
      { suffix: "RENEW2", trade: "trade-renew-2", money: 9900, hash: "hash-renew-2" },
    ]);
    assert.deepEqual(renewalResults.map((result) => result.rows[0].success), [true, true]);
    assert.equal(sql(`select count(*) from public.user_subscriptions where user_id='${ids.concurrentRenewalUser}'`), "2");
    assert.equal(sql(`select count(*) from (select starts_at,lag(ends_at) over(order by starts_at) previous_ends_at from public.user_subscriptions where user_id='${ids.concurrentRenewalUser}') s where starts_at=previous_ends_at`), "1");

    insertOrder({ suffix: "MIXED1", userId: ids.concurrentOverlapUser, productId: "00000000-0000-4000-8000-000000000902", code: "standard_monthly", money: 9900 });
    insertOrder({ suffix: "MIXED2", userId: ids.concurrentOverlapUser, productId: "00000000-0000-4000-8000-000000000903", code: "standard_yearly", money: 59900 });
    const overlapResults = await settleConcurrently([
      { suffix: "MIXED1", trade: "trade-mixed-1", money: 9900, hash: "hash-mixed-1" },
      { suffix: "MIXED2", trade: "trade-mixed-2", money: 59900, hash: "hash-mixed-2" },
    ]);
    assert.equal(overlapResults.filter((result) => result.rows[0].success).length, 1);
    assert.equal(overlapResults.filter((result) => result.rows[0].status === "overlapping_subscription").length, 1);
    assert.equal(sql(`select count(*) from public.user_subscriptions where user_id='${ids.concurrentOverlapUser}'`), "1");
    assert.equal(sql(`select count(*) from public.payment_orders where user_id='${ids.concurrentOverlapUser}' and grant_status='failed' and grant_error='overlapping_subscription'`), "1");
    sql(`drop trigger test_delay_concurrent_subscription_grant on public.user_subscriptions; drop function public.test_delay_concurrent_subscription_grant()`);

    const agenticCaseKey = "rectification:40000000-0000-4000-8000-000000000007";
    const agenticReservation = sql(`select reservation_id from public.authorize_usage('${ids.creditUser}','rectification',null,'${agenticCaseKey}',2)`);
    assert.equal(sql(`select reservation_id||':'||credits from public.authorize_usage('${ids.creditUser}','rectification',null,'${agenticCaseKey}',2)`), `${agenticReservation}:18`);
    assert.equal(sql(`select success||':'||reason from public.authorize_usage('${ids.creditUser}','report.full',null,'${agenticCaseKey}',2)`), "f:request_conflict");
    assert.equal(sql(`select success||':'||reason from public.authorize_usage('${ids.creditUser}','rectification','standard-model','${agenticCaseKey}',2)`), "f:request_conflict");
    assert.equal(sql(`select success||':'||reason from public.authorize_usage('${ids.creditUser}','rectification',null,'${agenticCaseKey}',3)`), "f:request_conflict");
    const firstTurnUsage = { eventKey: "turn-1", actualModelId: "standard-model", inputTokens: 10, outputTokens: 5, costMicrousd: 100, durationMs: 1000 };
    assert.equal(sql(`select success||':'||credits from public.complete_usage('${ids.creditUser}','${agenticCaseKey}','${sqlJson(firstTurnUsage)}'::jsonb)`), "true:18");
    assert.equal(sql(`select success||':'||credits from public.complete_usage('${ids.creditUser}','${agenticCaseKey}','${sqlJson(firstTurnUsage)}'::jsonb)`), "true:18");
    assert.equal(sql(`select success||':'||error_code from public.complete_usage('${ids.creditUser}','${agenticCaseKey}','${sqlJson({ ...firstTurnUsage, outputTokens: 6 })}'::jsonb)`), "f:event_payload_conflict");
    const concurrentUsage = await completeUsageConcurrently(ids.creditUser, agenticCaseKey, [
      { eventKey: "turn-2", actualModelId: "standard-model", inputTokens: 20, outputTokens: 10, costMicrousd: 200, durationMs: 2000 },
      { eventKey: "turn-3", actualModelId: "standard-model", inputTokens: 30, outputTokens: 15, costMicrousd: 300, durationMs: 3000 },
    ]);
    assert.deepEqual(concurrentUsage.map((result) => result.rows[0].success), [true, true]);
    assert.equal(sql(`select reservation_id||':'||credits from public.authorize_usage('${ids.creditUser}','rectification',null,'${agenticCaseKey}',2)`), `${agenticReservation}:18`);
    assert.equal(sql(`select count(*) from public.usage_reservations where user_id='${ids.creditUser}' and request_id='${agenticCaseKey}'`), "1");
    assert.equal(sql(`select count(*) from public.credit_transactions where user_id='${ids.creditUser}' and request_id='${agenticCaseKey}' and transaction_type='reserve'`), "1");
    assert.equal(sql(`select count(*) from public.usage_events where reservation_id='${agenticReservation}'`), "3");
    assert.equal(sql(`select input_tokens||':'||output_tokens||':'||cost_microusd||':'||duration_ms from public.usage_ledger where reservation_id='${agenticReservation}'`), "60:30:600:6000");

    sql(`update public.user_subscriptions set entitlement_snapshot='[{"featureKey":"chat.standard","allowanceType":"quota","allowanceCount":1,"resetPeriod":"billing_period","modelTier":"standard","metadata":{}}]'::jsonb where id='${monthlySubscription}'`);
    assert.equal(sql(`select source||':'||credits from public.authorize_usage('${ids.subscriber}','chat.standard',null,'quota-covered',1)`), "subscription:10");
    assert.equal(sql(`select source||':'||credits from public.authorize_usage('${ids.subscriber}','chat.standard',null,'quota-fallback',1)`), "credits:9");
    sql(`select * from public.release_usage('${ids.subscriber}','quota-covered','quota test cleanup')`);
    sql(`select * from public.release_usage('${ids.subscriber}','quota-fallback','quota test cleanup')`);

    sql(`update public.user_subscriptions set entitlement_snapshot='[{"featureKey":"chat.standard","allowanceType":"unlimited","allowanceCount":null,"resetPeriod":"billing_period","modelTier":"standard","metadata":{"minuteLimit":1}}]'::jsonb where id='${monthlySubscription}'`);
    assert.equal(sql(`select source||':'||credits from public.authorize_usage('${ids.subscriber}','chat.standard',null,'fair-covered',1)`), "subscription:10");
    assert.equal(sql(`select source||':'||credits from public.authorize_usage('${ids.subscriber}','chat.standard',null,'fair-fallback',1)`), "credits:9");
    sql(`select * from public.release_usage('${ids.subscriber}','fair-covered','fair use test cleanup')`);
    sql(`select * from public.release_usage('${ids.subscriber}','fair-fallback','fair use test cleanup')`);
    sql(`update public.user_subscriptions s set entitlement_snapshot=o.entitlement_snapshot from public.payment_orders o where s.id='${monthlySubscription}' and o.id=s.source_order_id`);

    assert.equal(sql(`select source||':'||credits from public.authorize_usage('${ids.subscriber}','chat.standard',null,'subscription-chat',1)`), "subscription:10");
    assert.equal(sql(`select success from public.complete_usage('${ids.subscriber}','subscription-chat','{"actualModelId":"standard-model","modelConfigVersion":7,"inputTokens":120,"outputTokens":45,"costMicrousd":321,"durationMs":900}'::jsonb)`), "t");
    assert.equal(sql(`select actual_model_id||':'||model_config_version||':'||input_tokens||':'||output_tokens||':'||cost_microusd from public.usage_ledger where user_id='${ids.subscriber}' and request_id='subscription-chat'`), "standard-model:7:120:45:321");
    assert.equal(sql(`select credits from public.profiles where id='${ids.subscriber}'`), "10");
    assert.equal(sql(`select error_code from public.release_usage('${ids.subscriber}','subscription-chat','late release')`), "request_completed");

    sql(`update public.feature_flags set enabled=false,rollout_percentage=0 where flag_key='billing.subscriptions' and status='published'`);
    assert.equal(sql(`select source||':'||credits from public.authorize_usage('${ids.subscriber}','chat.standard',null,'flag-off-chat',1)`), "credits:9");
    assert.equal(sql(`select success from public.complete_usage('${ids.subscriber}','flag-off-chat','{"eventKey":"flag-off-turn"}'::jsonb)`), "t");
    assert.equal(sql(`select count(*) from public.usage_reservations where request_id='flag-off-chat' and source='subscription'`), "0");
    assert.equal(sql(`select credits from public.profiles where id='${ids.subscriber}'`), "9");
    assert.equal(sql(`select success||':'||status from public.settle_order('${orderNo("SNAPSHOT")}','trade-snapshot',1200,'hash-snapshot-flag-off')`), "true:paid");
    assert.equal(sql(`select success||':'||status from public.settle_epay_order('${orderNo("SNAPSHOT")}','trade-snapshot',1200,'hash-snapshot-notify-flag-off')`), "true:paid");
    assert.equal(sql(`select count(*) from public.user_subscriptions where source_order_id=(select id from public.payment_orders where order_no='${orderNo("SNAPSHOT")}')`), "1");
    sql(`update public.profiles set credits=10 where id='${ids.subscriber}'`);
    sql(`update public.feature_flags set enabled=true,rollout_percentage=100 where flag_key='billing.subscriptions' and status='published'`);
    assert.equal(sql(`select source||':'||credits from public.authorize_usage('${ids.subscriber}','chat.standard',null,'flag-on-chat',1)`), "subscription:10");
    sql(`select * from public.release_usage('${ids.subscriber}','flag-on-chat','feature flag test cleanup')`);

    sql(`update public.user_subscriptions set starts_at=clock_timestamp()-interval '2 months',ends_at=clock_timestamp()-interval '1 second' where id='${monthlySubscription}'`);
    assert.equal(sql(`select source||':'||credits from public.authorize_usage('${ids.subscriber}','chat.standard',null,'credit-chat',1)`), "credits:9");
    assert.equal(sql(`select source||':'||credits from public.authorize_usage('${ids.subscriber}','chat.standard',null,'credit-chat',1)`), "credits:9");
    assert.equal(sql(`select count(*) from public.credit_transactions where user_id='${ids.subscriber}' and request_id='credit-chat' and transaction_type='reserve'`), "1");
    assert.equal(sql(`select success||':'||credits from public.release_usage('${ids.subscriber}','credit-chat','generation failed')`), "true:10");
    assert.equal(sql(`select count(*) from public.credit_transactions where user_id='${ids.subscriber}' and request_id='credit-chat' and transaction_type='refund'`), "1");

    sql(`update public.user_subscriptions set starts_at=clock_timestamp()-interval '1 day',ends_at=clock_timestamp()+interval '1 month' where id='${monthlySubscription}'`);
    const provider = sql(`select public.admin_save_model_provider('${ids.model}',null,'Test Provider','openai',null,'test-encrypted-api-key',true,'新增测试供应商','provider-1')`);
    const premiumDraft = sql(`select public.admin_save_model_draft('${ids.model}','premium-model',null,'${provider}','Premium','', 'premium-v1','premium',2,128000,100,200,true,true,null,'{}','新增高级模型','premium-draft-1')`);
    sql(`select public.admin_record_model_connection_test('${ids.model}','${premiumDraft}',200,'premium-test-1')`);
    sql(`select public.admin_publish_model('${ids.model}','${premiumDraft}','发布高级模型','premium-publish-1')`);
    assert.equal(sql(`select source||':'||credits from public.authorize_usage('${ids.subscriber}','chat.standard','premium-model','premium-fallback',2)`), "credits:8");
    sql(`select * from public.release_usage('${ids.subscriber}','premium-fallback','premium not covered')`);

    const currentEnds = sql(`select ends_at from public.user_subscriptions where id='${monthlySubscription}'`);
    sql(`select public.admin_adjust_subscription('${ids.billing}','${monthlySubscription}','extend',3,'${currentEnds}','人工补偿三天','adjust-1')`);
    expectSqlError(
      `select public.admin_adjust_subscription('${ids.billing}','${monthlySubscription}','extend',1,'${currentEnds}','使用过期版本调整','adjust-stale')`,
      /subscription_version_conflict/,
    );

    sql(`update public.model_config_versions set is_default=false where id='${premiumDraft}'`);
    const nonDefaultDraft = sql(`select public.admin_save_model_draft('${ids.model}','standard-model',null,'${provider}','Standard','', 'standard-v1','standard',1,64000,10,20,true,false,null,'{}','新增标准模型','standard-draft-1')`);
    sql(`select public.admin_record_model_connection_test('${ids.model}','${nonDefaultDraft}',200,'standard-test-invalid')`);
    expectSqlError(
      `select public.admin_publish_model('${ids.model}','${nonDefaultDraft}','无默认模型不得发布','standard-publish-invalid')`,
      /default_model_required/,
    );
    sql(`select public.admin_save_model_draft('${ids.model}','standard-model','${nonDefaultDraft}','${provider}','Standard','', 'standard-v1','standard',1,64000,10,20,true,true,null,'{}','设为默认模型','standard-draft-default')`);
    sql(`select public.admin_record_model_connection_test('${ids.model}','${nonDefaultDraft}',200,'standard-test-1')`);
    sql(`select public.admin_publish_model('${ids.model}','${nonDefaultDraft}','发布默认模型','standard-publish-1')`);
    assert.equal(sql(`select count(*) from public.model_config_versions where status='published' and enabled and is_default`), "1");

    const standardV2 = sql(`select public.admin_save_model_draft('${ids.model}','standard-model',null,'${provider}','Standard 2','', 'standard-v2','standard',1,64000,11,21,true,true,null,'{}','新建第二版','standard-draft-2')`);
    sql(`select public.admin_record_model_connection_test('${ids.model}','${standardV2}',200,'standard-test-2')`);
    sql(`select public.admin_publish_model('${ids.model}','${standardV2}','发布第二版','standard-publish-2')`);
    const standardConfig = sql("select id from public.model_configs where model_id='standard-model'");
    sql(`select public.admin_record_model_connection_test('${ids.model}','${nonDefaultDraft}',200,'standard-rollback-1')`);
    sql(`select public.admin_rollback_model('${ids.model}','${standardConfig}',1,'回滚第一版','standard-rollback-1')`);
    assert.equal(sql("select version from public.model_config_versions where config_id=(select id from public.model_configs where model_id='standard-model') and status='published'"), "1");

    const standardV3 = sql(`select public.admin_save_model_draft('${ids.model}','standard-model',null,'${provider}','Standard 3','', 'standard-v3','standard',1,64000,12,22,true,false,'premium-model','{}','设置回退到高级模型','standard-draft-3')`);
    sql(`select public.admin_record_model_connection_test('${ids.model}','${standardV3}',200,'standard-test-3')`);
    sql(`select public.admin_publish_model('${ids.model}','${standardV3}','发布第三版','standard-publish-3')`);
    const premiumV2 = sql(`select public.admin_save_model_draft('${ids.model}','premium-model',null,'${provider}','Premium 2','', 'premium-v2','premium',2,128000,101,201,true,true,'standard-model','{}','构造循环回退草稿','premium-draft-2')`);
    sql(`select public.admin_record_model_connection_test('${ids.model}','${premiumV2}',200,'premium-test-cycle')`);
    expectSqlError(
      `select public.admin_publish_model('${ids.model}','${premiumV2}','循环回退不得发布','premium-publish-cycle')`,
      /model_fallback_cycle/,
    );
    assert.equal(sql(`select count(*) from public.model_config_versions where status='published' and enabled and is_default`), "1");
    assert.equal(sql(`select count(*) from public.model_publish_events where action='rollback'`), "1");
  } finally {
    fixture.stop();
  }
});