create table if not exists assets (
  id text primary key,
  owner_id text not null,
  kind text not null,
  name text not null,
  url text not null,
  storage_path text,
  source text not null,
  tags text[] not null default '{}',
  metadata jsonb not null default '{}'::jsonb,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

create table if not exists generation_jobs (
  id text primary key,
  owner_id text not null,
  external_client_id text,
  capability text not null,
  provider text not null,
  req_key text not null,
  status text not null,
  prompt text,
  input_asset_ids text[] not null default '{}',
  input_urls text[] not null default '{}',
  output_asset_ids text[] not null default '{}',
  provider_task_id text,
  request_payload jsonb not null default '{}'::jsonb,
  response_payload jsonb,
  error jsonb,
  retry_of text,
  idempotency_key text,
  idempotency_fingerprint text,
  priority integer not null default 0,
  attempts integer not null default 0,
  max_attempts integer not null default 3,
  scheduled_at timestamptz not null default now(),
  locked_at timestamptz,
  locked_by text,
  started_at timestamptz,
  completed_at timestamptz,
  webhook_url text,
  webhook_attempts integer not null default 0,
  webhook_last_status jsonb,
  usage_context jsonb,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

alter table generation_jobs add column if not exists external_client_id text;
alter table generation_jobs add column if not exists idempotency_key text;
alter table generation_jobs add column if not exists idempotency_fingerprint text;
alter table generation_jobs add column if not exists priority integer not null default 0;
alter table generation_jobs add column if not exists attempts integer not null default 0;
alter table generation_jobs add column if not exists max_attempts integer not null default 3;
alter table generation_jobs add column if not exists scheduled_at timestamptz not null default now();
alter table generation_jobs add column if not exists locked_at timestamptz;
alter table generation_jobs add column if not exists locked_by text;
alter table generation_jobs add column if not exists started_at timestamptz;
alter table generation_jobs add column if not exists completed_at timestamptz;
alter table generation_jobs add column if not exists webhook_url text;
alter table generation_jobs add column if not exists webhook_attempts integer not null default 0;
alter table generation_jobs add column if not exists webhook_last_status jsonb;
alter table generation_jobs add column if not exists usage_context jsonb;

create table if not exists usage_events (
  id text primary key,
  owner_id text not null,
  job_id text not null,
  source text not null default 'platform',
  capability text not null,
  provider text,
  req_key text,
  account_username text,
  account_display_name text,
  tenant_id text,
  organization_id text,
  organization_name text,
  quantity integer not null default 1,
  estimated_unit text not null default 'job',
  created_at timestamptz not null default now()
);

alter table usage_events add column if not exists source text;
alter table usage_events add column if not exists provider text;
alter table usage_events add column if not exists req_key text;
alter table usage_events add column if not exists account_username text;
alter table usage_events add column if not exists account_display_name text;
alter table usage_events add column if not exists tenant_id text;
alter table usage_events add column if not exists organization_id text;
alter table usage_events add column if not exists organization_name text;
alter table usage_events add column if not exists quantity integer not null default 1;
alter table usage_events add column if not exists estimated_unit text not null default 'job';

update usage_events as usage
set source = case
      when jobs.external_client_id is not null or usage.owner_id like 'api:%' then 'api'
      else 'platform'
    end,
    provider = coalesce(usage.provider, jobs.provider),
    req_key = coalesce(usage.req_key, jobs.req_key),
    quantity = 1,
    estimated_unit = 'job'
from generation_jobs as jobs
where jobs.id = usage.job_id;

update usage_events
set source = case when owner_id like 'api:%' then 'api' else 'platform' end,
    quantity = 1,
    estimated_unit = 'job'
where source is null;

alter table usage_events alter column source set default 'platform';
alter table usage_events alter column source set not null;
alter table usage_events alter column estimated_unit set default 'job';
alter table usage_events drop constraint if exists usage_events_job_id_fkey;

create table if not exists projects (
  id text primary key,
  owner_id text not null,
  name text not null,
  brief text not null default '',
  type text not null default 'custom',
  asset_ids text[] not null default '{}',
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

create table if not exists image_templates (
  id text primary key,
  owner_id text not null,
  name text not null,
  description text,
  prompt text not null,
  preview_image_url text,
  settings jsonb not null default '{}'::jsonb,
  sort_order integer not null default 0,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

create index if not exists assets_owner_created_idx on assets(owner_id, created_at desc);
create index if not exists generation_jobs_owner_created_idx on generation_jobs(owner_id, created_at desc);
create index if not exists generation_jobs_status_idx on generation_jobs(status);
create index if not exists generation_jobs_claim_idx on generation_jobs(status, scheduled_at, locked_at, priority desc);
create index if not exists generation_jobs_external_client_idx on generation_jobs(owner_id, external_client_id, created_at desc);
create unique index if not exists generation_jobs_idempotency_idx
  on generation_jobs(owner_id, external_client_id, idempotency_key)
  where external_client_id is not null and idempotency_key is not null;
create index if not exists usage_events_owner_created_idx on usage_events(owner_id, created_at desc);
do $$
begin
  if exists (
    select 1
    from usage_events
    group by job_id
    having count(*) > 1
  ) then
    raise exception using
      errcode = '23505',
      message = 'USAGE_EVENTS_DUPLICATE_JOB_ID: back up and clean duplicate usage_events.job_id rows before retrying this migration';
  end if;
end;
$$;
create unique index if not exists usage_events_job_id_idx on usage_events(job_id);
create index if not exists usage_events_source_created_idx on usage_events(source, created_at desc);
create index if not exists usage_events_organization_created_idx on usage_events(organization_id, created_at desc);
create index if not exists image_templates_owner_sort_idx on image_templates(owner_id, sort_order asc, updated_at desc);

create or replace function claim_generation_jobs(
  p_worker_id text,
  p_limit integer default 1,
  p_lock_timeout_seconds integer default 300
)
returns setof generation_jobs
language plpgsql
set search_path = public, pg_temp
as $$
declare
  v_now timestamptz := now();
begin
  return query
  with candidates as (
    select id
    from generation_jobs
    where status in ('queued', 'running')
      and coalesce(scheduled_at, created_at) <= v_now
      and (
        locked_at is null
        or locked_at < v_now - make_interval(secs => p_lock_timeout_seconds)
      )
    order by coalesce(priority, 0) desc, coalesce(scheduled_at, created_at) asc, created_at asc
    limit greatest(1, least(coalesce(p_limit, 1), 20))
    for update skip locked
  ),
  updated as (
    update generation_jobs
    set locked_at = v_now,
        locked_by = p_worker_id,
        started_at = coalesce(generation_jobs.started_at, v_now),
        updated_at = v_now
    where id in (select id from candidates)
    returning generation_jobs.*
  )
  select * from updated;
end;
$$;

revoke all on function claim_generation_jobs(text, integer, integer) from public;

create table if not exists platform_organizations (
  id text primary key,
  name text not null unique,
  status text not null default 'active' check (status in ('active', 'disabled')),
  archive_owner_id text not null unique,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

create table if not exists platform_users (
  id text primary key,
  phone text not null unique,
  display_name text not null,
  role text not null default 'user' check (role in ('super_admin', 'organization_admin', 'user')),
  organization_id text references platform_organizations(id) on delete set null,
  status text not null default 'active' check (status in ('active', 'disabled')),
  password_hash text not null,
  password_salt text not null,
  failed_login_count integer not null default 0,
  locked_until timestamptz,
  session_version integer not null default 1,
  last_login_at timestamptz,
  legacy_subject text,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

create table if not exists platform_account_migrations (
  id text primary key,
  legacy_owner_id text not null unique,
  legacy_phone text,
  platform_user_id text not null,
  created_at timestamptz not null default now()
);

create index if not exists platform_users_organization_idx on platform_users(organization_id, created_at desc);
create index if not exists platform_users_status_idx on platform_users(status, role);
create index if not exists platform_account_migrations_user_idx on platform_account_migrations(platform_user_id);

alter table generation_jobs add column if not exists billing jsonb;
alter table usage_events add column if not exists charged_amount_fen bigint;
alter table usage_events add column if not exists currency text;

create table if not exists billing_price_rules (
  id text primary key,
  provider text not null check (provider in ('volcengine-visual', 'evolink', 'seedance', 'bailian', 'mock')),
  capability text not null check (capability in ('image.generate', 'video.generate')),
  req_key text,
  variant_key text,
  unit text not null check (unit in ('request', 'image', 'video_second')),
  standard_unit_price_fen bigint not null check (standard_unit_price_fen >= 0),
  markup_multiplier numeric(12, 4) not null default 1.0000 check (markup_multiplier >= 1),
  enabled boolean not null default true,
  conditions jsonb not null default '{}'::jsonb,
  quantity_source text check (quantity_source in ('request', 'image_count', 'duration')),
  priority integer not null default 0,
  note text,
  source jsonb,
  parameter_dimensions jsonb not null default '[]'::jsonb,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

alter table billing_price_rules add column if not exists variant_key text;
alter table billing_price_rules add column if not exists source jsonb;
alter table billing_price_rules add column if not exists conditions jsonb not null default '{}'::jsonb;
alter table billing_price_rules add column if not exists quantity_source text;
alter table billing_price_rules add column if not exists priority integer not null default 0;
alter table billing_price_rules add column if not exists parameter_dimensions jsonb not null default '[]'::jsonb;
alter table billing_price_rules drop constraint if exists billing_price_rules_quantity_source_check;
alter table billing_price_rules add constraint billing_price_rules_quantity_source_check check (quantity_source is null or quantity_source in ('request', 'image_count', 'duration'));

drop index if exists billing_price_rules_match_idx;
create unique index if not exists billing_price_rules_match_idx
  on billing_price_rules(provider, capability, coalesce(req_key, ''), coalesce(variant_key, ''), coalesce(conditions, '{}'::jsonb));

create table if not exists billing_wallets (
  organization_id text primary key references platform_organizations(id) on delete restrict,
  balance_fen bigint not null default 0 check (balance_fen >= 0),
  total_recharged_fen bigint not null default 0 check (total_recharged_fen >= 0),
  total_charged_fen bigint not null default 0 check (total_charged_fen >= 0),
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

create table if not exists billing_ledger (
  id text primary key,
  organization_id text not null references platform_organizations(id) on delete restrict,
  account_id text,
  job_id text,
  kind text not null check (kind in ('recharge', 'charge', 'refund', 'adjustment')),
  delta_fen bigint not null check (delta_fen <> 0),
  balance_after_fen bigint not null check (balance_after_fen >= 0),
  currency text not null default 'CNY' check (currency = 'CNY'),
  idempotency_key text not null,
  description text not null,
  metadata jsonb not null default '{}'::jsonb,
  created_at timestamptz not null default now()
);

do $$
declare
  v_constraint record;
  v_index record;
begin
  for v_constraint in
    select constraint_schema, constraint_name
    from information_schema.table_constraints
    where table_schema = current_schema()
      and table_name = 'billing_ledger'
      and constraint_type = 'UNIQUE'
      and array(
        select key_column_usage.column_name::text
        from information_schema.key_column_usage
        where key_column_usage.constraint_schema = table_constraints.constraint_schema
          and key_column_usage.constraint_name = table_constraints.constraint_name
          and key_column_usage.table_name = table_constraints.table_name
        order by key_column_usage.ordinal_position
      ) = array['idempotency_key']::text[]
  loop
    execute format('alter table %I.%I drop constraint %I', current_schema(), 'billing_ledger', v_constraint.constraint_name);
  end loop;

  for v_index in
    select indexes.schemaname, indexes.indexname
    from pg_indexes as indexes
    join pg_class as index_class on index_class.relname = indexes.indexname
    join pg_namespace as index_namespace
      on index_namespace.oid = index_class.relnamespace
     and index_namespace.nspname = indexes.schemaname
    join pg_index as index_metadata on index_metadata.indexrelid = index_class.oid
    join pg_attribute as indexed_column
      on indexed_column.attrelid = index_metadata.indrelid
     and indexed_column.attnum = index_metadata.indkey[0]
    left join pg_constraint as backing_constraint on backing_constraint.conindid = index_class.oid
    where indexes.schemaname = current_schema()
      and indexes.tablename = 'billing_ledger'
      and index_metadata.indisunique
      and index_metadata.indnkeyatts = 1
      and indexed_column.attname = 'idempotency_key'
      and backing_constraint.oid is null
  loop
    execute format('drop index %I.%I', v_index.schemaname, v_index.indexname);
  end loop;
end;
$$;

create unique index if not exists billing_ledger_organization_idempotency_idx
  on billing_ledger(organization_id, idempotency_key);
create index if not exists billing_ledger_organization_created_idx on billing_ledger(organization_id, created_at desc);
create index if not exists billing_ledger_account_created_idx on billing_ledger(account_id, created_at desc);
create index if not exists billing_ledger_job_idx on billing_ledger(job_id);

create or replace function billing_post_wallet_entry(
  p_ledger_id text,
  p_organization_id text,
  p_account_id text,
  p_job_id text,
  p_kind text,
  p_delta_fen bigint,
  p_currency text,
  p_idempotency_key text,
  p_description text,
  p_metadata jsonb
)
returns table (
  ledger_id text,
  balance_after_fen bigint,
  balance_fen bigint,
  total_recharged_fen bigint,
  total_charged_fen bigint,
  created_at timestamptz,
  updated_at timestamptz,
  delta_fen bigint
)
language plpgsql
set search_path = public, pg_temp
as $$
declare
  v_existing billing_ledger%rowtype;
  v_wallet billing_wallets%rowtype;
  v_entry billing_ledger%rowtype;
  v_account_id text;
begin
  if p_currency <> 'CNY' then
    raise exception 'BILLING_UNSUPPORTED_CURRENCY';
  end if;
  if p_delta_fen = 0 then
    raise exception 'BILLING_ZERO_DELTA';
  end if;

  v_account_id := case when p_kind in ('recharge', 'adjustment') then null else p_account_id end;

  perform pg_advisory_xact_lock(
    hashtextextended(jsonb_build_array(p_organization_id, p_idempotency_key)::text, 0)
  );

  select * into v_existing
  from billing_ledger
  where organization_id = p_organization_id
    and idempotency_key = p_idempotency_key;
  if found then
    if v_existing.account_id is distinct from v_account_id
      or v_existing.job_id is distinct from p_job_id
      or v_existing.kind is distinct from p_kind
      or v_existing.delta_fen is distinct from p_delta_fen
      or v_existing.currency is distinct from p_currency
    then
      raise exception using
        errcode = 'P0001',
        message = 'BILLING_IDEMPOTENCY_PAYLOAD_MISMATCH';
    end if;

    select * into v_wallet from billing_wallets where organization_id = p_organization_id;
    return query select v_existing.id, v_existing.balance_after_fen, v_wallet.balance_fen,
      v_wallet.total_recharged_fen, v_wallet.total_charged_fen, v_existing.created_at,
      v_wallet.updated_at, v_existing.delta_fen;
    return;
  end if;

  insert into billing_wallets(organization_id)
  values (p_organization_id)
  on conflict (organization_id) do nothing;

  select * into v_wallet from billing_wallets
  where organization_id = p_organization_id
  for update;

  if p_delta_fen < 0 and v_wallet.balance_fen < abs(p_delta_fen) then
    raise exception 'BILLING_INSUFFICIENT_BALANCE';
  end if;

  update billing_wallets
  set balance_fen = v_wallet.balance_fen + p_delta_fen,
      total_recharged_fen = v_wallet.total_recharged_fen + case when p_kind = 'recharge' and p_delta_fen > 0 then p_delta_fen else 0 end,
      total_charged_fen = v_wallet.total_charged_fen + case when p_kind = 'charge' and p_delta_fen < 0 then abs(p_delta_fen) else 0 end,
      updated_at = now()
  where organization_id = p_organization_id
  returning * into v_wallet;

  insert into billing_ledger(
    id, organization_id, account_id, job_id, kind, delta_fen,
    balance_after_fen, currency, idempotency_key, description, metadata
  ) values (
    p_ledger_id, p_organization_id,
    v_account_id,
    p_job_id, p_kind, p_delta_fen,
    v_wallet.balance_fen, p_currency, p_idempotency_key, p_description, coalesce(p_metadata, '{}'::jsonb)
  ) returning * into v_entry;

  return query select v_entry.id, v_entry.balance_after_fen, v_wallet.balance_fen,
    v_wallet.total_recharged_fen, v_wallet.total_charged_fen, v_entry.created_at,
    v_wallet.updated_at, v_entry.delta_fen;
end;
$$;

revoke all on function billing_post_wallet_entry(text, text, text, text, text, bigint, text, text, text, jsonb) from public;