From c2fd1400bed8dc808785f4a66dab3c3758398006 Mon Sep 17 00:00:00 2001 From: xiaohe Date: Mon, 20 Jul 2026 19:10:32 +0800 Subject: [PATCH 1/4] feat(db): persist lowest product price candles --- scripts/verify-supabase-recovery-baseline.mjs | 3 + src/lib/admin.ts | 6 +- .../20260720220000_product_price_history.sql | 684 +++++++++++++++++ supabase/recovery-baseline.json | 4 +- supabase/schema.sql | 685 ++++++++++++++++++ 5 files changed, 1379 insertions(+), 3 deletions(-) create mode 100644 supabase/migrations/20260720220000_product_price_history.sql diff --git a/scripts/verify-supabase-recovery-baseline.mjs b/scripts/verify-supabase-recovery-baseline.mjs index 9b779f0..36a8d7a 100644 --- a/scripts/verify-supabase-recovery-baseline.mjs +++ b/scripts/verify-supabase-recovery-baseline.mjs @@ -32,6 +32,9 @@ assert( for (const requiredSql of [ "create table if not exists canonical_products", "create table if not exists raw_offers", + "create table if not exists product_price_samples", + "create table if not exists product_price_candles", + "create or replace function record_product_price_samples", "create table if not exists public_user_profiles", "create or replace function public.claim_runtime_lease", "create or replace function public.consume_feedback_evidence_upload_quota", diff --git a/src/lib/admin.ts b/src/lib/admin.ts index f2c1538..9f8210f 100644 --- a/src/lib/admin.ts +++ b/src/lib/admin.ts @@ -584,9 +584,12 @@ export async function upsertRawOffer(input: OfferInput & { sourceId?: string | n failureReason: existingManualHidden?.failureReason, }; - const { error } = await supabase.from("raw_offers").upsert(toRawOfferRow(offer)); + const row = toRawOfferRow(offer); + const { error } = await supabase.from("raw_offers").upsert(row); if (error) throw error; + await upsertRawOfferConfirmations([row]); + return offer; } @@ -914,6 +917,7 @@ function rawOfferConfirmationRow(row: Record): Record= 0); + end if; +end $$; + +do $$ +begin + if not exists ( + select 1 + from pg_constraint + where conname = 'raw_offer_confirmations_canonical_product_id_fkey' + and conrelid = 'raw_offer_confirmations'::regclass + ) then + alter table raw_offer_confirmations + add constraint raw_offer_confirmations_canonical_product_id_fkey + foreign key (canonical_product_id) + references canonical_products(id) + on delete set null; + end if; +end $$; + +update raw_offer_confirmations confirmations +set + canonical_product_id = offers.canonical_product_id, + consecutive_valid_confirmations = case + when offers.canonical_product_id is not null + and confirmations.price > 0 + and confirmations.source_status in ('in_stock', 'low_stock') + and confirmations.effective_status = 'available' + and confirmations.freshness_status = 'fresh' + and (confirmations.stock_count is null or confirmations.stock_count > 0) + and (confirmations.expires_at is null or confirmations.expires_at > confirmations.confirmed_at) + then 1 + else 0 + end +from raw_offers offers +where offers.id = confirmations.raw_offer_id; + +create or replace function priceai_track_offer_confirmation_streak() +returns trigger +language plpgsql +set search_path = public +as $$ +declare + raw_offer raw_offers%rowtype; + is_valid boolean := false; +begin + select * into raw_offer + from raw_offers + where id = new.raw_offer_id; + + is_valid := + found + and raw_offer.hidden = false + and raw_offer.canonical_product_id is not null + and raw_offer.canonical_product_id is not distinct from new.canonical_product_id + and raw_offer.price is not distinct from new.price + and raw_offer.price > 0 + and raw_offer.currency = 'CNY' + and trim(raw_offer.url) <> '' + and raw_offer.status in ('in_stock', 'low_stock') + and raw_offer.source_status in ('in_stock', 'low_stock') + and raw_offer.effective_status = 'available' + and raw_offer.freshness_status = 'fresh' + and (raw_offer.stock_count is null or raw_offer.stock_count > 0) + and coalesce(raw_offer.min_order_quantity, 1) <= 1 + and not ( + coalesce(raw_offer.public_filter_tags, '{}'::text[]) + && array['shared_access', 'web_only_account', 'domestic_mirror_site']::text[] + ) + and new.price > 0 + and new.source_status in ('in_stock', 'low_stock') + and new.effective_status = 'available' + and new.freshness_status = 'fresh' + and (new.stock_count is null or new.stock_count > 0) + and (new.expires_at is null or new.expires_at > greatest(new.confirmed_at, now())); + + if tg_op = 'UPDATE' and new.confirmed_at <= old.confirmed_at then + new := old; + if pg_trigger_depth() > 1 and not is_valid then + new.consecutive_valid_confirmations := 0; + end if; + return new; + end if; + + if not is_valid then + new.consecutive_valid_confirmations := 0; + elsif tg_op = 'INSERT' then + new.consecutive_valid_confirmations := 1; + elsif old.consecutive_valid_confirmations > 0 + and new.canonical_product_id is not distinct from old.canonical_product_id + and new.price is not distinct from old.price + then + new.consecutive_valid_confirmations := least(old.consecutive_valid_confirmations + 1, 32767); + else + new.consecutive_valid_confirmations := 1; + end if; + + return new; +end; +$$; + +create or replace function priceai_reset_offer_confirmation_streak() +returns trigger +language plpgsql +set search_path = public +as $$ +begin + update raw_offer_confirmations + set + source_status = new.source_status, + effective_status = new.effective_status, + freshness_status = new.freshness_status, + stock_count = new.stock_count, + expires_at = new.expires_at, + updated_at = now() + where raw_offer_id = new.id; + return new; +end; +$$; + +drop trigger if exists raw_offer_confirmations_track_streak on raw_offer_confirmations; +create trigger raw_offer_confirmations_track_streak +before insert or update on raw_offer_confirmations +for each row execute function priceai_track_offer_confirmation_streak(); + +drop trigger if exists raw_offers_reset_confirmation_streak on raw_offers; +create trigger raw_offers_reset_confirmation_streak +after update of hidden, canonical_product_id, source_title, price, currency, status, source_status, url, + tags, stock_count, min_order_quantity, effective_status, + freshness_status, expires_at +on raw_offers +for each row execute function priceai_reset_offer_confirmation_streak(); + +revoke all on function priceai_track_offer_confirmation_streak() from public; +revoke all on function priceai_reset_offer_confirmation_streak() from public; + +create index if not exists raw_offer_confirmations_product_streak_idx + on raw_offer_confirmations(canonical_product_id, consecutive_valid_confirmations) + where consecutive_valid_confirmations >= 2; + +create table if not exists product_price_samples ( + product_id text not null references canonical_products(id) on delete cascade, + bucket_start timestamptz not null, + observed_at timestamptz not null, + price numeric not null check (price > 0), + currency text not null default 'CNY' check (currency = 'CNY'), + offer_id text references raw_offers(id) on delete set null, + source_id text references sources(id) on delete set null, + eligible_offer_count integer not null check (eligible_offer_count > 0), + sample_method text not null default 'lowest_valid_available_offer' + check (sample_method = 'lowest_valid_available_offer'), + created_at timestamptz not null default now(), + updated_at timestamptz not null default now(), + primary key (product_id, bucket_start) +); + +comment on table product_price_samples is + 'Fifteen-minute observations of each standard product lowest eligible single-unit CNY offer.'; + +create index if not exists product_price_samples_product_observed_idx + on product_price_samples(product_id, observed_at desc); +create index if not exists product_price_samples_observed_idx + on product_price_samples(observed_at); + +create table if not exists product_price_candles ( + product_id text not null references canonical_products(id) on delete cascade, + candle_interval text not null check (candle_interval in ('1h', '1d')), + bucket_start timestamptz not null, + bucket_end timestamptz not null, + open_price numeric not null check (open_price > 0), + high_price numeric not null check (high_price > 0), + low_price numeric not null check (low_price > 0), + close_price numeric not null check (close_price > 0), + currency text not null default 'CNY' check (currency = 'CNY'), + sample_count integer not null check (sample_count > 0), + eligible_offer_count integer not null check (eligible_offer_count > 0), + first_sample_at timestamptz not null, + last_sample_at timestamptz not null, + created_at timestamptz not null default now(), + updated_at timestamptz not null default now(), + primary key (product_id, candle_interval, bucket_start), + check (bucket_end > bucket_start), + check (high_price >= greatest(open_price, close_price)), + check (low_price <= least(open_price, close_price)) +); + +create index if not exists product_price_candles_interval_bucket_idx + on product_price_candles(candle_interval, bucket_start desc, product_id); + +alter table product_price_samples enable row level security; +alter table product_price_candles enable row level security; +revoke all on table product_price_samples from anon, authenticated, public; +revoke all on table product_price_candles from anon, authenticated, public; +grant select, insert, update, delete on table product_price_samples to service_role; +grant select, insert, update, delete on table product_price_candles to service_role; + +drop trigger if exists product_price_samples_set_updated_at on product_price_samples; +create trigger product_price_samples_set_updated_at +before update on product_price_samples +for each row execute function set_updated_at(); + +drop trigger if exists product_price_candles_set_updated_at on product_price_candles; +create trigger product_price_candles_set_updated_at +before update on product_price_candles +for each row execute function set_updated_at(); + +create or replace view product_price_eligible_offers +with (security_invoker = true) as +select + offers.id as offer_id, + offers.source_id, + offers.canonical_product_id as product_id, + offers.price, + offers.currency, + coalesce( + confirmations.verified_at, + confirmations.last_seen_at, + confirmations.captured_at, + offers.verified_at, + offers.captured_at, + offers.source_updated_at + ) as public_updated_at +from raw_offers offers +join raw_offer_confirmations confirmations + on confirmations.raw_offer_id = offers.id +where offers.canonical_product_id is not null + and confirmations.canonical_product_id = offers.canonical_product_id + and confirmations.price = offers.price + and confirmations.consecutive_valid_confirmations >= 2 + and offers.hidden = false + and offers.price is not null + and offers.price > 0 + and offers.currency = 'CNY' + and trim(offers.url) <> '' + and offers.status in ('in_stock', 'low_stock') + and confirmations.source_status in ('in_stock', 'low_stock') + and (confirmations.stock_count is null or confirmations.stock_count > 0) + and confirmations.effective_status = 'available' + and confirmations.freshness_status = 'fresh' + and (confirmations.expires_at is null or confirmations.expires_at > now()) + and coalesce(offers.min_order_quantity, 1) <= 1 + and not ( + coalesce(offers.public_filter_tags, '{}'::text[]) + && array['shared_access', 'web_only_account', 'domestic_mirror_site']::text[] + ); + +revoke all on table product_price_eligible_offers from anon, authenticated, public; +grant select on table product_price_eligible_offers to service_role; + +create or replace function record_product_price_samples( + p_product_ids text[] default null, + p_observed_at timestamptz default now() +) +returns jsonb +language plpgsql +security definer +set search_path = public +as $$ +declare + observed_at_value timestamptz := coalesce(p_observed_at, now()); + sample_bucket timestamptz := to_timestamp( + floor(extract(epoch from coalesce(p_observed_at, now())) / 900) * 900 + ); + target_product_count integer := 0; + sampled_product_ids text[] := '{}'::text[]; + sampled_count integer := 0; +begin + select count(*) + into target_product_count + from canonical_products products + where products.is_active = true + and (p_product_ids is null or products.id = any(p_product_ids)); + + with active_products as ( + select products.id + from canonical_products products + where products.is_active = true + and (p_product_ids is null or products.id = any(p_product_ids)) + ), + ranked as ( + select + eligible.*, + count(*) over (partition by eligible.product_id) as eligible_offer_count, + row_number() over ( + partition by eligible.product_id + order by + eligible.price asc, + eligible.public_updated_at desc nulls last, + eligible.offer_id asc + ) as price_rank + from product_price_eligible_offers eligible + join active_products products on products.id = eligible.product_id + ), + inserted as ( + insert into product_price_samples ( + product_id, + bucket_start, + observed_at, + price, + currency, + offer_id, + source_id, + eligible_offer_count, + sample_method + ) + select + ranked.product_id, + sample_bucket, + observed_at_value, + ranked.price, + ranked.currency, + ranked.offer_id, + ranked.source_id, + ranked.eligible_offer_count::integer, + 'lowest_valid_available_offer' + from ranked + where ranked.price_rank = 1 + on conflict (product_id, bucket_start) do update + set + observed_at = excluded.observed_at, + price = excluded.price, + currency = excluded.currency, + offer_id = excluded.offer_id, + source_id = excluded.source_id, + eligible_offer_count = excluded.eligible_offer_count, + sample_method = excluded.sample_method + where product_price_samples.observed_at < excluded.observed_at + returning product_id, bucket_start, observed_at + ) + select coalesce(array_agg(inserted.product_id), '{}'::text[]) + into sampled_product_ids + from inserted; + + sampled_count := coalesce(array_length(sampled_product_ids, 1), 0); + + if sampled_count > 0 then + with affected_samples as ( + select distinct + inserted.product_id, + inserted.observed_at + from ( + select samples.product_id, samples.observed_at + from product_price_samples samples + where samples.product_id = any(sampled_product_ids) + and samples.bucket_start = sample_bucket + ) inserted + ), + target_buckets as ( + select distinct + affected_samples.product_id, + intervals.candle_interval, + case + when intervals.candle_interval = '1d' then + date_trunc('day', affected_samples.observed_at at time zone 'Asia/Shanghai') at time zone 'Asia/Shanghai' + else date_trunc('hour', affected_samples.observed_at) + end as bucket_start + from affected_samples + cross join (values ('1h'), ('1d')) as intervals(candle_interval) + ), + aggregated as ( + select + target_buckets.product_id, + target_buckets.candle_interval, + target_buckets.bucket_start, + target_buckets.bucket_start + case + when target_buckets.candle_interval = '1d' then interval '1 day' + else interval '1 hour' + end as bucket_end, + (array_agg(samples.price order by samples.observed_at asc))[1] as open_price, + max(samples.price) as high_price, + min(samples.price) as low_price, + (array_agg(samples.price order by samples.observed_at desc))[1] as close_price, + count(*)::integer as sample_count, + (array_agg(samples.eligible_offer_count order by samples.observed_at desc))[1] + as eligible_offer_count, + min(samples.observed_at) as first_sample_at, + max(samples.observed_at) as last_sample_at + from target_buckets + join product_price_samples samples + on samples.product_id = target_buckets.product_id + and samples.observed_at >= target_buckets.bucket_start + and samples.observed_at < target_buckets.bucket_start + case + when target_buckets.candle_interval = '1d' then interval '1 day' + else interval '1 hour' + end + group by + target_buckets.product_id, + target_buckets.candle_interval, + target_buckets.bucket_start + ) + insert into product_price_candles ( + product_id, + candle_interval, + bucket_start, + bucket_end, + open_price, + high_price, + low_price, + close_price, + currency, + sample_count, + eligible_offer_count, + first_sample_at, + last_sample_at + ) + select + aggregated.product_id, + aggregated.candle_interval, + aggregated.bucket_start, + aggregated.bucket_end, + aggregated.open_price, + aggregated.high_price, + aggregated.low_price, + aggregated.close_price, + 'CNY', + aggregated.sample_count, + aggregated.eligible_offer_count, + aggregated.first_sample_at, + aggregated.last_sample_at + from aggregated + on conflict (product_id, candle_interval, bucket_start) do update + set + bucket_end = excluded.bucket_end, + open_price = excluded.open_price, + high_price = excluded.high_price, + low_price = excluded.low_price, + close_price = excluded.close_price, + currency = excluded.currency, + sample_count = excluded.sample_count, + eligible_offer_count = excluded.eligible_offer_count, + first_sample_at = excluded.first_sample_at, + last_sample_at = excluded.last_sample_at; + end if; + + return jsonb_build_object( + 'productsProcessed', target_product_count, + 'samplesWritten', sampled_count, + 'productsWithoutPrice', greatest(target_product_count - sampled_count, 0), + 'observedAt', observed_at_value, + 'bucketStart', sample_bucket + ); +end; +$$; + +revoke execute on function record_product_price_samples(text[], timestamptz) + from anon, authenticated, public; +grant execute on function record_product_price_samples(text[], timestamptz) + to service_role; + +create or replace function list_product_price_current(p_product_ids text[]) +returns table ( + product_id text, + current_price numeric, + currency text, + eligible_offer_count integer, + quoted_at timestamptz +) +language sql +stable +security definer +set search_path = public +as $$ + with ranked as ( + select + eligible.*, + count(*) over (partition by eligible.product_id) as offer_count, + row_number() over ( + partition by eligible.product_id + order by + eligible.price asc, + eligible.public_updated_at desc nulls last, + eligible.offer_id asc + ) as price_rank + from product_price_eligible_offers eligible + where eligible.product_id = any(coalesce(p_product_ids, '{}'::text[])) + ) + select + ranked.product_id, + ranked.price, + ranked.currency, + ranked.offer_count::integer, + ranked.public_updated_at + from ranked + where ranked.price_rank = 1 + order by ranked.product_id; +$$; + +revoke execute on function list_product_price_current(text[]) + from anon, authenticated, public; +grant execute on function list_product_price_current(text[]) + to service_role; + +create or replace function list_public_product_price_candles( + p_product_ids text[], + p_interval text default '1d', + p_limit_per_product integer default 90, + p_before timestamptz default null +) +returns table ( + product_id text, + period_start timestamptz, + open_price numeric, + high_price numeric, + low_price numeric, + close_price numeric, + currency text, + sample_count integer, + eligible_offer_count integer, + first_sample_at timestamptz, + last_sample_at timestamptz +) +language sql +stable +security definer +set search_path = public +as $$ + with ranked as ( + select + candles.*, + row_number() over ( + partition by candles.product_id + order by candles.bucket_start desc + ) as recent_rank + from product_price_candles candles + where candles.product_id = any(coalesce(p_product_ids, '{}'::text[])) + and candles.candle_interval = p_interval + and p_interval in ('1h', '1d') + and (p_before is null or candles.bucket_start < p_before) + ) + select + ranked.product_id, + ranked.bucket_start, + ranked.open_price, + ranked.high_price, + ranked.low_price, + ranked.close_price, + ranked.currency, + ranked.sample_count, + ranked.eligible_offer_count, + ranked.first_sample_at, + ranked.last_sample_at + from ranked + where ranked.recent_rank <= greatest( + 1, + least( + coalesce(p_limit_per_product, case when p_interval = '1h' then 168 else 90 end), + case when p_interval = '1h' then 720 else 365 end + ) + ) + order by ranked.product_id, ranked.bucket_start asc; +$$; + +revoke execute on function list_public_product_price_candles(text[], text, integer, timestamptz) + from anon, authenticated, public; +grant execute on function list_public_product_price_candles(text[], text, integer, timestamptz) + to service_role; + +create or replace function prune_product_price_history( + p_batch_size integer default 5000, + p_dry_run boolean default true +) +returns jsonb +language plpgsql +security definer +set search_path = public +as $$ +declare + batch_size_value integer := greatest(100, least(coalesce(p_batch_size, 5000), 20000)); + sample_candidates integer := 0; + hourly_candidates integer := 0; + deleted_samples integer := 0; + deleted_hourly_candles integer := 0; +begin + select count(*) + into sample_candidates + from ( + select samples.product_id, samples.bucket_start + from product_price_samples samples + where samples.observed_at < now() - interval '180 days' + and exists ( + select 1 + from product_price_candles hourly + where hourly.product_id = samples.product_id + and hourly.candle_interval = '1h' + and hourly.bucket_start = date_trunc('hour', samples.observed_at) + ) + and exists ( + select 1 + from product_price_candles daily + where daily.product_id = samples.product_id + and daily.candle_interval = '1d' + and daily.bucket_start = + date_trunc('day', samples.observed_at at time zone 'Asia/Shanghai') at time zone 'Asia/Shanghai' + ) + order by samples.observed_at + limit batch_size_value + ) candidates; + + select count(*) + into hourly_candidates + from ( + select candles.product_id, candles.bucket_start + from product_price_candles candles + where candles.candle_interval = '1h' + and candles.bucket_start < now() - interval '365 days' + order by candles.bucket_start + limit batch_size_value + ) candidates; + + if not p_dry_run then + with expired as ( + select samples.product_id, samples.bucket_start + from product_price_samples samples + where samples.observed_at < now() - interval '180 days' + and exists ( + select 1 + from product_price_candles hourly + where hourly.product_id = samples.product_id + and hourly.candle_interval = '1h' + and hourly.bucket_start = date_trunc('hour', samples.observed_at) + ) + and exists ( + select 1 + from product_price_candles daily + where daily.product_id = samples.product_id + and daily.candle_interval = '1d' + and daily.bucket_start = + date_trunc('day', samples.observed_at at time zone 'Asia/Shanghai') at time zone 'Asia/Shanghai' + ) + order by samples.observed_at + limit batch_size_value + ) + delete from product_price_samples samples + using expired + where samples.product_id = expired.product_id + and samples.bucket_start = expired.bucket_start; + get diagnostics deleted_samples = row_count; + + with expired as ( + select candles.product_id, candles.candle_interval, candles.bucket_start + from product_price_candles candles + where candles.candle_interval = '1h' + and candles.bucket_start < now() - interval '365 days' + order by candles.bucket_start + limit batch_size_value + ) + delete from product_price_candles candles + using expired + where candles.product_id = expired.product_id + and candles.candle_interval = expired.candle_interval + and candles.bucket_start = expired.bucket_start; + get diagnostics deleted_hourly_candles = row_count; + end if; + + return jsonb_build_object( + 'dryRun', p_dry_run, + 'batchSize', batch_size_value, + 'sampleCandidates', sample_candidates, + 'hourlyCandleCandidates', hourly_candidates, + 'deletedSamples', deleted_samples, + 'deletedHourlyCandles', deleted_hourly_candles + ); +end; +$$; + +revoke execute on function prune_product_price_history(integer, boolean) + from anon, authenticated, public; +grant execute on function prune_product_price_history(integer, boolean) + to service_role; diff --git a/supabase/recovery-baseline.json b/supabase/recovery-baseline.json index 280298b..15f3b00 100644 --- a/supabase/recovery-baseline.json +++ b/supabase/recovery-baseline.json @@ -1,8 +1,8 @@ { "version": 1, "schemaPath": "supabase/schema.sql", - "schemaSha256": "f3ce8992eec4113970085fed3ec1474a401cf92d629be55ab39485a1f4fd9591", + "schemaSha256": "0c1ff6d8378a8a05e34b51269b15f8bf3dcc534d61ffcf57affddfbf8b6f5778", "includesMigrationsThrough": "20260721133000", - "includedMigrationCount": 140, + "includedMigrationCount": 141, "createdAt": "2026-07-21T13:30:00.000Z" } diff --git a/supabase/schema.sql b/supabase/schema.sql index 8a717cb..b5e61e4 100644 --- a/supabase/schema.sql +++ b/supabase/schema.sql @@ -4569,3 +4569,688 @@ revoke execute on function finalize_channel_submission_approval(text, text, json grant execute on function priceai_channel_submission_key(text) to service_role; grant execute on function list_submission_price_benchmarks(text[]) to service_role; grant execute on function finalize_channel_submission_approval(text, text, jsonb, timestamptz, jsonb, text) to service_role; + +alter table raw_offer_confirmations + add column if not exists canonical_product_id text; +alter table raw_offer_confirmations + add column if not exists consecutive_valid_confirmations integer not null default 0; + +do $$ +begin + if not exists ( + select 1 + from pg_constraint + where conname = 'raw_offer_confirmations_consecutive_valid_check' + and conrelid = 'raw_offer_confirmations'::regclass + ) then + alter table raw_offer_confirmations + add constraint raw_offer_confirmations_consecutive_valid_check + check (consecutive_valid_confirmations >= 0); + end if; +end $$; + +do $$ +begin + if not exists ( + select 1 + from pg_constraint + where conname = 'raw_offer_confirmations_canonical_product_id_fkey' + and conrelid = 'raw_offer_confirmations'::regclass + ) then + alter table raw_offer_confirmations + add constraint raw_offer_confirmations_canonical_product_id_fkey + foreign key (canonical_product_id) + references canonical_products(id) + on delete set null; + end if; +end $$; + +update raw_offer_confirmations confirmations +set + canonical_product_id = offers.canonical_product_id, + consecutive_valid_confirmations = case + when offers.canonical_product_id is not null + and confirmations.price > 0 + and confirmations.source_status in ('in_stock', 'low_stock') + and confirmations.effective_status = 'available' + and confirmations.freshness_status = 'fresh' + and (confirmations.stock_count is null or confirmations.stock_count > 0) + and (confirmations.expires_at is null or confirmations.expires_at > confirmations.confirmed_at) + then 1 + else 0 + end +from raw_offers offers +where offers.id = confirmations.raw_offer_id; + +create or replace function priceai_track_offer_confirmation_streak() +returns trigger +language plpgsql +set search_path = public +as $$ +declare + raw_offer raw_offers%rowtype; + is_valid boolean := false; +begin + select * into raw_offer + from raw_offers + where id = new.raw_offer_id; + + is_valid := + found + and raw_offer.hidden = false + and raw_offer.canonical_product_id is not null + and raw_offer.canonical_product_id is not distinct from new.canonical_product_id + and raw_offer.price is not distinct from new.price + and raw_offer.price > 0 + and raw_offer.currency = 'CNY' + and trim(raw_offer.url) <> '' + and raw_offer.status in ('in_stock', 'low_stock') + and raw_offer.source_status in ('in_stock', 'low_stock') + and raw_offer.effective_status = 'available' + and raw_offer.freshness_status = 'fresh' + and (raw_offer.stock_count is null or raw_offer.stock_count > 0) + and coalesce(raw_offer.min_order_quantity, 1) <= 1 + and not ( + coalesce(raw_offer.public_filter_tags, '{}'::text[]) + && array['shared_access', 'web_only_account', 'domestic_mirror_site']::text[] + ) + and new.price > 0 + and new.source_status in ('in_stock', 'low_stock') + and new.effective_status = 'available' + and new.freshness_status = 'fresh' + and (new.stock_count is null or new.stock_count > 0) + and (new.expires_at is null or new.expires_at > greatest(new.confirmed_at, now())); + + if tg_op = 'UPDATE' and new.confirmed_at <= old.confirmed_at then + new := old; + if pg_trigger_depth() > 1 and not is_valid then + new.consecutive_valid_confirmations := 0; + end if; + return new; + end if; + + if not is_valid then + new.consecutive_valid_confirmations := 0; + elsif tg_op = 'INSERT' then + new.consecutive_valid_confirmations := 1; + elsif old.consecutive_valid_confirmations > 0 + and new.canonical_product_id is not distinct from old.canonical_product_id + and new.price is not distinct from old.price + then + new.consecutive_valid_confirmations := least(old.consecutive_valid_confirmations + 1, 32767); + else + new.consecutive_valid_confirmations := 1; + end if; + + return new; +end; +$$; + +create or replace function priceai_reset_offer_confirmation_streak() +returns trigger +language plpgsql +set search_path = public +as $$ +begin + update raw_offer_confirmations + set + source_status = new.source_status, + effective_status = new.effective_status, + freshness_status = new.freshness_status, + stock_count = new.stock_count, + expires_at = new.expires_at, + updated_at = now() + where raw_offer_id = new.id; + return new; +end; +$$; + +drop trigger if exists raw_offer_confirmations_track_streak on raw_offer_confirmations; +create trigger raw_offer_confirmations_track_streak +before insert or update on raw_offer_confirmations +for each row execute function priceai_track_offer_confirmation_streak(); + +drop trigger if exists raw_offers_reset_confirmation_streak on raw_offers; +create trigger raw_offers_reset_confirmation_streak +after update of hidden, canonical_product_id, source_title, price, currency, status, source_status, url, + tags, stock_count, min_order_quantity, effective_status, + freshness_status, expires_at +on raw_offers +for each row execute function priceai_reset_offer_confirmation_streak(); + +revoke all on function priceai_track_offer_confirmation_streak() from public; +revoke all on function priceai_reset_offer_confirmation_streak() from public; + +create index if not exists raw_offer_confirmations_product_streak_idx + on raw_offer_confirmations(canonical_product_id, consecutive_valid_confirmations) + where consecutive_valid_confirmations >= 2; + +create table if not exists product_price_samples ( + product_id text not null references canonical_products(id) on delete cascade, + bucket_start timestamptz not null, + observed_at timestamptz not null, + price numeric not null check (price > 0), + currency text not null default 'CNY' check (currency = 'CNY'), + offer_id text references raw_offers(id) on delete set null, + source_id text references sources(id) on delete set null, + eligible_offer_count integer not null check (eligible_offer_count > 0), + sample_method text not null default 'lowest_valid_available_offer' + check (sample_method = 'lowest_valid_available_offer'), + created_at timestamptz not null default now(), + updated_at timestamptz not null default now(), + primary key (product_id, bucket_start) +); + +comment on table product_price_samples is + 'Fifteen-minute observations of each standard product lowest eligible single-unit CNY offer.'; + +create index if not exists product_price_samples_product_observed_idx + on product_price_samples(product_id, observed_at desc); +create index if not exists product_price_samples_observed_idx + on product_price_samples(observed_at); + +create table if not exists product_price_candles ( + product_id text not null references canonical_products(id) on delete cascade, + candle_interval text not null check (candle_interval in ('1h', '1d')), + bucket_start timestamptz not null, + bucket_end timestamptz not null, + open_price numeric not null check (open_price > 0), + high_price numeric not null check (high_price > 0), + low_price numeric not null check (low_price > 0), + close_price numeric not null check (close_price > 0), + currency text not null default 'CNY' check (currency = 'CNY'), + sample_count integer not null check (sample_count > 0), + eligible_offer_count integer not null check (eligible_offer_count > 0), + first_sample_at timestamptz not null, + last_sample_at timestamptz not null, + created_at timestamptz not null default now(), + updated_at timestamptz not null default now(), + primary key (product_id, candle_interval, bucket_start), + check (bucket_end > bucket_start), + check (high_price >= greatest(open_price, close_price)), + check (low_price <= least(open_price, close_price)) +); + +create index if not exists product_price_candles_interval_bucket_idx + on product_price_candles(candle_interval, bucket_start desc, product_id); + +alter table product_price_samples enable row level security; +alter table product_price_candles enable row level security; +revoke all on table product_price_samples from anon, authenticated, public; +revoke all on table product_price_candles from anon, authenticated, public; +grant select, insert, update, delete on table product_price_samples to service_role; +grant select, insert, update, delete on table product_price_candles to service_role; + +drop trigger if exists product_price_samples_set_updated_at on product_price_samples; +create trigger product_price_samples_set_updated_at +before update on product_price_samples +for each row execute function set_updated_at(); + +drop trigger if exists product_price_candles_set_updated_at on product_price_candles; +create trigger product_price_candles_set_updated_at +before update on product_price_candles +for each row execute function set_updated_at(); + +create or replace view product_price_eligible_offers +with (security_invoker = true) as +select + offers.id as offer_id, + offers.source_id, + offers.canonical_product_id as product_id, + offers.price, + offers.currency, + coalesce( + confirmations.verified_at, + confirmations.last_seen_at, + confirmations.captured_at, + offers.verified_at, + offers.captured_at, + offers.source_updated_at + ) as public_updated_at +from raw_offers offers +join raw_offer_confirmations confirmations + on confirmations.raw_offer_id = offers.id +where offers.canonical_product_id is not null + and confirmations.canonical_product_id = offers.canonical_product_id + and confirmations.price = offers.price + and confirmations.consecutive_valid_confirmations >= 2 + and offers.hidden = false + and offers.price is not null + and offers.price > 0 + and offers.currency = 'CNY' + and trim(offers.url) <> '' + and offers.status in ('in_stock', 'low_stock') + and confirmations.source_status in ('in_stock', 'low_stock') + and (confirmations.stock_count is null or confirmations.stock_count > 0) + and confirmations.effective_status = 'available' + and confirmations.freshness_status = 'fresh' + and (confirmations.expires_at is null or confirmations.expires_at > now()) + and coalesce(offers.min_order_quantity, 1) <= 1 + and not ( + coalesce(offers.public_filter_tags, '{}'::text[]) + && array['shared_access', 'web_only_account', 'domestic_mirror_site']::text[] + ); + +revoke all on table product_price_eligible_offers from anon, authenticated, public; +grant select on table product_price_eligible_offers to service_role; + +create or replace function record_product_price_samples( + p_product_ids text[] default null, + p_observed_at timestamptz default now() +) +returns jsonb +language plpgsql +security definer +set search_path = public +as $$ +declare + observed_at_value timestamptz := coalesce(p_observed_at, now()); + sample_bucket timestamptz := to_timestamp( + floor(extract(epoch from coalesce(p_observed_at, now())) / 900) * 900 + ); + target_product_count integer := 0; + sampled_product_ids text[] := '{}'::text[]; + sampled_count integer := 0; +begin + select count(*) + into target_product_count + from canonical_products products + where products.is_active = true + and (p_product_ids is null or products.id = any(p_product_ids)); + + with active_products as ( + select products.id + from canonical_products products + where products.is_active = true + and (p_product_ids is null or products.id = any(p_product_ids)) + ), + ranked as ( + select + eligible.*, + count(*) over (partition by eligible.product_id) as eligible_offer_count, + row_number() over ( + partition by eligible.product_id + order by + eligible.price asc, + eligible.public_updated_at desc nulls last, + eligible.offer_id asc + ) as price_rank + from product_price_eligible_offers eligible + join active_products products on products.id = eligible.product_id + ), + inserted as ( + insert into product_price_samples ( + product_id, + bucket_start, + observed_at, + price, + currency, + offer_id, + source_id, + eligible_offer_count, + sample_method + ) + select + ranked.product_id, + sample_bucket, + observed_at_value, + ranked.price, + ranked.currency, + ranked.offer_id, + ranked.source_id, + ranked.eligible_offer_count::integer, + 'lowest_valid_available_offer' + from ranked + where ranked.price_rank = 1 + on conflict (product_id, bucket_start) do update + set + observed_at = excluded.observed_at, + price = excluded.price, + currency = excluded.currency, + offer_id = excluded.offer_id, + source_id = excluded.source_id, + eligible_offer_count = excluded.eligible_offer_count, + sample_method = excluded.sample_method + where product_price_samples.observed_at < excluded.observed_at + returning product_id, bucket_start, observed_at + ) + select coalesce(array_agg(inserted.product_id), '{}'::text[]) + into sampled_product_ids + from inserted; + + sampled_count := coalesce(array_length(sampled_product_ids, 1), 0); + + if sampled_count > 0 then + with affected_samples as ( + select distinct + inserted.product_id, + inserted.observed_at + from ( + select samples.product_id, samples.observed_at + from product_price_samples samples + where samples.product_id = any(sampled_product_ids) + and samples.bucket_start = sample_bucket + ) inserted + ), + target_buckets as ( + select distinct + affected_samples.product_id, + intervals.candle_interval, + case + when intervals.candle_interval = '1d' then + date_trunc('day', affected_samples.observed_at at time zone 'Asia/Shanghai') at time zone 'Asia/Shanghai' + else date_trunc('hour', affected_samples.observed_at) + end as bucket_start + from affected_samples + cross join (values ('1h'), ('1d')) as intervals(candle_interval) + ), + aggregated as ( + select + target_buckets.product_id, + target_buckets.candle_interval, + target_buckets.bucket_start, + target_buckets.bucket_start + case + when target_buckets.candle_interval = '1d' then interval '1 day' + else interval '1 hour' + end as bucket_end, + (array_agg(samples.price order by samples.observed_at asc))[1] as open_price, + max(samples.price) as high_price, + min(samples.price) as low_price, + (array_agg(samples.price order by samples.observed_at desc))[1] as close_price, + count(*)::integer as sample_count, + (array_agg(samples.eligible_offer_count order by samples.observed_at desc))[1] + as eligible_offer_count, + min(samples.observed_at) as first_sample_at, + max(samples.observed_at) as last_sample_at + from target_buckets + join product_price_samples samples + on samples.product_id = target_buckets.product_id + and samples.observed_at >= target_buckets.bucket_start + and samples.observed_at < target_buckets.bucket_start + case + when target_buckets.candle_interval = '1d' then interval '1 day' + else interval '1 hour' + end + group by + target_buckets.product_id, + target_buckets.candle_interval, + target_buckets.bucket_start + ) + insert into product_price_candles ( + product_id, + candle_interval, + bucket_start, + bucket_end, + open_price, + high_price, + low_price, + close_price, + currency, + sample_count, + eligible_offer_count, + first_sample_at, + last_sample_at + ) + select + aggregated.product_id, + aggregated.candle_interval, + aggregated.bucket_start, + aggregated.bucket_end, + aggregated.open_price, + aggregated.high_price, + aggregated.low_price, + aggregated.close_price, + 'CNY', + aggregated.sample_count, + aggregated.eligible_offer_count, + aggregated.first_sample_at, + aggregated.last_sample_at + from aggregated + on conflict (product_id, candle_interval, bucket_start) do update + set + bucket_end = excluded.bucket_end, + open_price = excluded.open_price, + high_price = excluded.high_price, + low_price = excluded.low_price, + close_price = excluded.close_price, + currency = excluded.currency, + sample_count = excluded.sample_count, + eligible_offer_count = excluded.eligible_offer_count, + first_sample_at = excluded.first_sample_at, + last_sample_at = excluded.last_sample_at; + end if; + + return jsonb_build_object( + 'productsProcessed', target_product_count, + 'samplesWritten', sampled_count, + 'productsWithoutPrice', greatest(target_product_count - sampled_count, 0), + 'observedAt', observed_at_value, + 'bucketStart', sample_bucket + ); +end; +$$; + +revoke execute on function record_product_price_samples(text[], timestamptz) + from anon, authenticated, public; +grant execute on function record_product_price_samples(text[], timestamptz) + to service_role; + +create or replace function list_product_price_current(p_product_ids text[]) +returns table ( + product_id text, + current_price numeric, + currency text, + eligible_offer_count integer, + quoted_at timestamptz +) +language sql +stable +security definer +set search_path = public +as $$ + with ranked as ( + select + eligible.*, + count(*) over (partition by eligible.product_id) as offer_count, + row_number() over ( + partition by eligible.product_id + order by + eligible.price asc, + eligible.public_updated_at desc nulls last, + eligible.offer_id asc + ) as price_rank + from product_price_eligible_offers eligible + where eligible.product_id = any(coalesce(p_product_ids, '{}'::text[])) + ) + select + ranked.product_id, + ranked.price, + ranked.currency, + ranked.offer_count::integer, + ranked.public_updated_at + from ranked + where ranked.price_rank = 1 + order by ranked.product_id; +$$; + +revoke execute on function list_product_price_current(text[]) + from anon, authenticated, public; +grant execute on function list_product_price_current(text[]) + to service_role; + +create or replace function list_public_product_price_candles( + p_product_ids text[], + p_interval text default '1d', + p_limit_per_product integer default 90, + p_before timestamptz default null +) +returns table ( + product_id text, + period_start timestamptz, + open_price numeric, + high_price numeric, + low_price numeric, + close_price numeric, + currency text, + sample_count integer, + eligible_offer_count integer, + first_sample_at timestamptz, + last_sample_at timestamptz +) +language sql +stable +security definer +set search_path = public +as $$ + with ranked as ( + select + candles.*, + row_number() over ( + partition by candles.product_id + order by candles.bucket_start desc + ) as recent_rank + from product_price_candles candles + where candles.product_id = any(coalesce(p_product_ids, '{}'::text[])) + and candles.candle_interval = p_interval + and p_interval in ('1h', '1d') + and (p_before is null or candles.bucket_start < p_before) + ) + select + ranked.product_id, + ranked.bucket_start, + ranked.open_price, + ranked.high_price, + ranked.low_price, + ranked.close_price, + ranked.currency, + ranked.sample_count, + ranked.eligible_offer_count, + ranked.first_sample_at, + ranked.last_sample_at + from ranked + where ranked.recent_rank <= greatest( + 1, + least( + coalesce(p_limit_per_product, case when p_interval = '1h' then 168 else 90 end), + case when p_interval = '1h' then 720 else 365 end + ) + ) + order by ranked.product_id, ranked.bucket_start asc; +$$; + +revoke execute on function list_public_product_price_candles(text[], text, integer, timestamptz) + from anon, authenticated, public; +grant execute on function list_public_product_price_candles(text[], text, integer, timestamptz) + to service_role; + +create or replace function prune_product_price_history( + p_batch_size integer default 5000, + p_dry_run boolean default true +) +returns jsonb +language plpgsql +security definer +set search_path = public +as $$ +declare + batch_size_value integer := greatest(100, least(coalesce(p_batch_size, 5000), 20000)); + sample_candidates integer := 0; + hourly_candidates integer := 0; + deleted_samples integer := 0; + deleted_hourly_candles integer := 0; +begin + select count(*) + into sample_candidates + from ( + select samples.product_id, samples.bucket_start + from product_price_samples samples + where samples.observed_at < now() - interval '180 days' + and exists ( + select 1 + from product_price_candles hourly + where hourly.product_id = samples.product_id + and hourly.candle_interval = '1h' + and hourly.bucket_start = date_trunc('hour', samples.observed_at) + ) + and exists ( + select 1 + from product_price_candles daily + where daily.product_id = samples.product_id + and daily.candle_interval = '1d' + and daily.bucket_start = + date_trunc('day', samples.observed_at at time zone 'Asia/Shanghai') at time zone 'Asia/Shanghai' + ) + order by samples.observed_at + limit batch_size_value + ) candidates; + + select count(*) + into hourly_candidates + from ( + select candles.product_id, candles.bucket_start + from product_price_candles candles + where candles.candle_interval = '1h' + and candles.bucket_start < now() - interval '365 days' + order by candles.bucket_start + limit batch_size_value + ) candidates; + + if not p_dry_run then + with expired as ( + select samples.product_id, samples.bucket_start + from product_price_samples samples + where samples.observed_at < now() - interval '180 days' + and exists ( + select 1 + from product_price_candles hourly + where hourly.product_id = samples.product_id + and hourly.candle_interval = '1h' + and hourly.bucket_start = date_trunc('hour', samples.observed_at) + ) + and exists ( + select 1 + from product_price_candles daily + where daily.product_id = samples.product_id + and daily.candle_interval = '1d' + and daily.bucket_start = + date_trunc('day', samples.observed_at at time zone 'Asia/Shanghai') at time zone 'Asia/Shanghai' + ) + order by samples.observed_at + limit batch_size_value + ) + delete from product_price_samples samples + using expired + where samples.product_id = expired.product_id + and samples.bucket_start = expired.bucket_start; + get diagnostics deleted_samples = row_count; + + with expired as ( + select candles.product_id, candles.candle_interval, candles.bucket_start + from product_price_candles candles + where candles.candle_interval = '1h' + and candles.bucket_start < now() - interval '365 days' + order by candles.bucket_start + limit batch_size_value + ) + delete from product_price_candles candles + using expired + where candles.product_id = expired.product_id + and candles.candle_interval = expired.candle_interval + and candles.bucket_start = expired.bucket_start; + get diagnostics deleted_hourly_candles = row_count; + end if; + + return jsonb_build_object( + 'dryRun', p_dry_run, + 'batchSize', batch_size_value, + 'sampleCandidates', sample_candidates, + 'hourlyCandleCandidates', hourly_candidates, + 'deletedSamples', deleted_samples, + 'deletedHourlyCandles', deleted_hourly_candles + ); +end; +$$; + +revoke execute on function prune_product_price_history(integer, boolean) + from anon, authenticated, public; +grant execute on function prune_product_price_history(integer, boolean) + to service_role; From dbdd23782bf8bc2c8b92472330f2118d486299d7 Mon Sep 17 00:00:00 2001 From: xiaohe Date: Mon, 20 Jul 2026 19:11:15 +0800 Subject: [PATCH 2/4] feat(api): expose product price history --- .../collect-product-price-history.yml | 84 +++++++ .github/workflows/maintenance-retention.yml | 60 +++++ .../api/cron/product-price-history/route.ts | 58 +++++ src/app/api/price-chart-summaries/route.ts | 44 ++++ .../api/products/[id]/price-candles/route.ts | 56 +++++ src/lib/price-history-db.ts | 216 ++++++++++++++++++ src/lib/price-history.ts | 174 ++++++++++++++ 7 files changed, 692 insertions(+) create mode 100644 .github/workflows/collect-product-price-history.yml create mode 100644 src/app/api/cron/product-price-history/route.ts create mode 100644 src/app/api/price-chart-summaries/route.ts create mode 100644 src/app/api/products/[id]/price-candles/route.ts create mode 100644 src/lib/price-history-db.ts create mode 100644 src/lib/price-history.ts diff --git a/.github/workflows/collect-product-price-history.yml b/.github/workflows/collect-product-price-history.yml new file mode 100644 index 0000000..cf304b2 --- /dev/null +++ b/.github/workflows/collect-product-price-history.yml @@ -0,0 +1,84 @@ +name: Collect Product Price History + +on: + schedule: + - cron: "*/15 * * * *" + workflow_dispatch: + +concurrency: + group: collect-product-price-history + cancel-in-progress: false + +jobs: + sample: + runs-on: ubuntu-latest + timeout-minutes: 5 + steps: + - name: Record lowest available product prices + env: + COLLECT_PRICES_URL: ${{ secrets.COLLECT_PRICES_URL }} + CRON_SECRET: ${{ secrets.CRON_SECRET }} + run: | + set -euo pipefail + + if [ -z "$COLLECT_PRICES_URL" ] || [ -z "$CRON_SECRET" ]; then + echo "缺少 COLLECT_PRICES_URL 或 CRON_SECRET secret" + exit 1 + fi + + BASE_URL=$(node <<'NODE' + const url = new URL(process.env.COLLECT_PRICES_URL); + url.hash = ""; + url.search = ""; + url.pathname = url.pathname.replace(/\/api\/cron\/[^/]+\/?$/, "") || "/"; + console.log(url.toString().replace(/\/$/, "")); + NODE + ) + + SAMPLE_URL="$BASE_URL/api/cron/product-price-history?action=sample" node <<'NODE' + const response = await fetch(process.env.SAMPLE_URL, { + method: "POST", + headers: { authorization: `Bearer ${process.env.CRON_SECRET}` }, + }); + const text = await response.text(); + let payload = null; + try { + payload = text ? JSON.parse(text) : null; + } catch { + payload = null; + } + + if (!response.ok || payload?.ok === false) { + throw new Error(payload?.message || text || `HTTP ${response.status}`); + } + + console.log(JSON.stringify({ + action: payload.action, + result: payload.result, + startedAt: payload.startedAt, + finishedAt: payload.finishedAt, + })); + NODE + + - name: Notify failure + if: failure() + env: + WEBHOOK_URL: ${{ secrets.PRICEAI_ALERT_WEBHOOK_URL }} + run: | + if [ -z "$WEBHOOK_URL" ]; then + exit 0 + fi + + node <<'NODE' + await fetch(process.env.WEBHOOK_URL, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + title: "PriceAI 商品价格历史采样失败", + message: `${process.env.GITHUB_WORKFLOW} #${process.env.GITHUB_RUN_NUMBER} failed on ${process.env.GITHUB_REF_NAME}`, + severity: "warning", + source: "github-actions", + runUrl: `${process.env.GITHUB_SERVER_URL}/${process.env.GITHUB_REPOSITORY}/actions/runs/${process.env.GITHUB_RUN_ID}`, + }), + }); + NODE diff --git a/.github/workflows/maintenance-retention.yml b/.github/workflows/maintenance-retention.yml index 5d86432..41599e0 100644 --- a/.github/workflows/maintenance-retention.yml +++ b/.github/workflows/maintenance-retention.yml @@ -84,6 +84,66 @@ jobs: })); NODE + - name: Apply product price history retention + env: + COLLECT_PRICES_URL: ${{ secrets.COLLECT_PRICES_URL }} + CRON_SECRET: ${{ secrets.CRON_SECRET }} + INPUT_BATCH_SIZE: ${{ inputs.batch_size }} + INPUT_DRY_RUN: ${{ inputs.dry_run }} + run: | + set -euo pipefail + + if [ -z "$COLLECT_PRICES_URL" ] || [ -z "$CRON_SECRET" ]; then + echo "缺少 COLLECT_PRICES_URL 或 CRON_SECRET secret" + exit 1 + fi + + BASE_URL=$(node <<'NODE' + const url = new URL(process.env.COLLECT_PRICES_URL); + url.hash = ""; + url.search = ""; + url.pathname = url.pathname.replace(/\/api\/cron\/[^/]+\/?$/, "") || "/"; + console.log(url.toString().replace(/\/$/, "")); + NODE + ) + + BATCH_SIZE="${INPUT_BATCH_SIZE:-5000}" + if ! [[ "$BATCH_SIZE" =~ ^[0-9]+$ ]] || [ "$BATCH_SIZE" -lt 100 ] || [ "$BATCH_SIZE" -gt 20000 ]; then + echo "batch_size 必须是 100 到 20000 之间的整数" + exit 1 + fi + + APPLY=1 + if [ "${INPUT_DRY_RUN:-false}" = "true" ]; then + APPLY=0 + fi + + RETENTION_URL="$BASE_URL/api/cron/product-price-history?action=prune&apply=$APPLY&batch=$BATCH_SIZE" node <<'NODE' + const response = await fetch(process.env.RETENTION_URL, { + method: "POST", + headers: { authorization: `Bearer ${process.env.CRON_SECRET}` }, + }); + const text = await response.text(); + let payload = null; + try { + payload = text ? JSON.parse(text) : null; + } catch { + payload = null; + } + + if (!response.ok || payload?.ok === false) { + throw new Error(payload?.message || text || `HTTP ${response.status}`); + } + + console.log(JSON.stringify({ + dryRun: payload.dryRun, + batchSize: payload.batchSize, + result: payload.result, + startedAt: payload.startedAt, + finishedAt: payload.finishedAt, + })); + NODE + - name: Notify failure if: failure() env: diff --git a/src/app/api/cron/product-price-history/route.ts b/src/app/api/cron/product-price-history/route.ts new file mode 100644 index 0000000..6fa663f --- /dev/null +++ b/src/app/api/cron/product-price-history/route.ts @@ -0,0 +1,58 @@ +import { logApiError, safeApiErrorMessage } from "@/lib/api-errors"; +import { authorizeCronRequest, cronMethodNotAllowed } from "@/lib/cron-auth"; +import { pruneProductPriceHistory, recordProductPriceSamples } from "@/lib/price-history-db"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; +export const maxDuration = 120; + +export function GET() { + return cronMethodNotAllowed("执行标准商品价格历史维护"); +} + +export async function POST(request: Request) { + const authError = authorizeCronRequest(request, "执行标准商品价格历史维护"); + if (authError) return authError; + + const url = new URL(request.url); + const action = url.searchParams.get("action"); + if (action !== "sample" && action !== "prune") { + return Response.json({ ok: false, message: "action 仅支持 sample 或 prune。" }, { status: 400 }); + } + + const startedAt = new Date().toISOString(); + const batchSize = boundedBatchSize(url.searchParams.get("batch")); + if (action === "prune" && batchSize === null) { + return Response.json({ ok: false, message: "batch 必须是 100 到 20000 之间的整数。" }, { status: 400 }); + } + + try { + const dryRun = url.searchParams.get("apply") !== "1"; + const result = action === "sample" + ? await recordProductPriceSamples(startedAt) + : await pruneProductPriceHistory({ batchSize: batchSize || 5000, dryRun }); + return Response.json({ + ok: true, + action, + startedAt, + finishedAt: new Date().toISOString(), + ...(action === "prune" ? { dryRun, batchSize } : {}), + result, + }, { headers: { "Cache-Control": "no-store" } }); + } catch (error) { + logApiError(`cron product price history ${action}`, error); + return Response.json({ + ok: false, + action, + startedAt, + finishedAt: new Date().toISOString(), + message: safeApiErrorMessage(error, "标准商品价格历史维护失败。"), + }, { status: 500, headers: { "Cache-Control": "no-store" } }); + } +} + +function boundedBatchSize(value: string | null): number | null { + if (!value) return 5000; + const parsed = Number(value); + return Number.isInteger(parsed) && parsed >= 100 && parsed <= 20000 ? parsed : null; +} diff --git a/src/app/api/price-chart-summaries/route.ts b/src/app/api/price-chart-summaries/route.ts new file mode 100644 index 0000000..6b6aea6 --- /dev/null +++ b/src/app/api/price-chart-summaries/route.ts @@ -0,0 +1,44 @@ +import { NextRequest, NextResponse } from "next/server"; +import { publicPriceApiErrorResponse } from "@/lib/api-errors"; +import { priceDataCacheHeaders } from "@/lib/cache-headers"; +import { cacheSearchParams } from "@/lib/cloudflare-cache-key"; +import { withCloudflarePublicCache } from "@/lib/cloudflare-edge-cache"; +import { parsePriceChartPoints, parsePriceHistoryInterval } from "@/lib/price-history"; +import { getProductPriceChartSummaries } from "@/lib/price-history-db"; +import { PRICE_DATA_EDGE_SECONDS } from "@/lib/public-cache-policy"; + +export const dynamic = "force-dynamic"; +export const revalidate = 0; + +export async function GET(request: NextRequest) { + const interval = parsePriceHistoryInterval(request.nextUrl.searchParams.get("interval") || "1d"); + if (!interval) return NextResponse.json({ message: "K 线周期仅支持 1h 或 1d。" }, { status: 400 }); + const points = parsePriceChartPoints(interval, request.nextUrl.searchParams.get("points")); + if (!points) return NextResponse.json({ message: "points 仅支持 24 或 30。" }, { status: 400 }); + + const platform = normalizedFilter(request.nextUrl.searchParams.get("platform")); + const productType = normalizedFilter(request.nextUrl.searchParams.get("productType")); + if (platform === null || productType === null) { + return NextResponse.json({ message: "筛选参数不能超过 80 个字符。" }, { status: 400 }); + } + + return withCloudflarePublicCache(request, { + namespace: "price-chart-summaries-v1", + ttlSeconds: PRICE_DATA_EDGE_SECONDS, + cacheKeySearchParams: cacheSearchParams({ interval, points, platform, productType }), + load: async () => { + try { + const result = await getProductPriceChartSummaries({ interval, points, platform, productType }); + return NextResponse.json(result, { headers: priceDataCacheHeaders() }); + } catch (error) { + return publicPriceApiErrorResponse("public price chart summaries API", error); + } + }, + }); +} + +function normalizedFilter(value: string | null): string | null | undefined { + const normalized = value?.trim(); + if (!normalized || normalized === "全部") return undefined; + return normalized.length <= 80 ? normalized : null; +} diff --git a/src/app/api/products/[id]/price-candles/route.ts b/src/app/api/products/[id]/price-candles/route.ts new file mode 100644 index 0000000..c7d6fe2 --- /dev/null +++ b/src/app/api/products/[id]/price-candles/route.ts @@ -0,0 +1,56 @@ +import { NextRequest, NextResponse } from "next/server"; +import { publicPriceApiErrorResponse } from "@/lib/api-errors"; +import { priceDataCacheHeaders } from "@/lib/cache-headers"; +import { cacheSearchParams } from "@/lib/cloudflare-cache-key"; +import { withCloudflarePublicCache } from "@/lib/cloudflare-edge-cache"; +import { + parseExclusiveBefore, + parsePriceHistoryInterval, + parsePriceHistoryLimit, +} from "@/lib/price-history"; +import { getProductPriceCandles, resolvePriceHistoryProduct } from "@/lib/price-history-db"; +import { PRICE_DATA_EDGE_SECONDS } from "@/lib/public-cache-policy"; + +export const dynamic = "force-dynamic"; +export const revalidate = 0; + +export async function GET( + request: NextRequest, + { params }: { params: Promise<{ id: string }> }, +) { + const { id } = await params; + const product = resolvePriceHistoryProduct(id); + if (!product) return NextResponse.json({ message: "标准商品不存在。" }, { status: 404 }); + + const interval = parsePriceHistoryInterval(request.nextUrl.searchParams.get("interval") || "1d"); + if (!interval) return NextResponse.json({ message: "K 线周期仅支持 1h 或 1d。" }, { status: 400 }); + + const limit = parsePriceHistoryLimit(interval, request.nextUrl.searchParams.get("limit")); + if (limit === null) { + return NextResponse.json( + { message: `limit 必须是 1 到 ${interval === "1h" ? 720 : 365} 之间的整数。` }, + { status: 400 }, + ); + } + const before = parseExclusiveBefore(request.nextUrl.searchParams.get("before")); + if (before === null) return NextResponse.json({ message: "before 必须是带时区的 ISO 时间。" }, { status: 400 }); + + return withCloudflarePublicCache(request, { + namespace: "product-price-candles-v1", + ttlSeconds: PRICE_DATA_EDGE_SECONDS, + cacheKeySearchParams: cacheSearchParams({ interval, limit, before }), + load: async () => { + try { + const result = await getProductPriceCandles({ + productId: product.id, + interval, + limit, + before, + }); + return NextResponse.json(result, { headers: priceDataCacheHeaders() }); + } catch (error) { + return publicPriceApiErrorResponse("public product price candles API", error); + } + }, + }); +} diff --git a/src/lib/price-history-db.ts b/src/lib/price-history-db.ts new file mode 100644 index 0000000..c455433 --- /dev/null +++ b/src/lib/price-history-db.ts @@ -0,0 +1,216 @@ +import "server-only"; + +import { publicCatalogProducts } from "./catalog"; +import { + compactProductPriceCandle, + groupProductPriceCandleRows, + productPriceChange, + type PriceHistoryInterval, + type ProductPriceCandleResponse, + type ProductPriceChartSummaryResponse, + type ProductPriceCurrent, +} from "./price-history"; +import { getSupabaseServerClient } from "./supabase"; + +type CurrentPriceRow = { + product_id?: unknown; + current_price?: unknown; + eligible_offer_count?: unknown; + quoted_at?: unknown; +}; + +export function resolvePriceHistoryProduct(productKey: string) { + return publicCatalogProducts().find((product) => product.id === productKey || product.slug === productKey) || null; +} + +export async function getProductPriceCandles(input: { + productId: string; + interval: PriceHistoryInterval; + limit: number; + before?: string; +}): Promise { + const generatedAt = new Date().toISOString(); + const supabase = getSupabaseServerClient(); + if (!supabase) return emptyCandleResponse(input.productId, input.interval, generatedAt); + + const [candlesResult, latestCandlesResult, currentResult] = await Promise.all([ + supabase.rpc("list_public_product_price_candles", { + p_product_ids: [input.productId], + p_interval: input.interval, + p_limit_per_product: input.limit, + p_before: input.before || null, + }), + supabase.rpc("list_public_product_price_candles", { + p_product_ids: [input.productId], + p_interval: input.interval, + p_limit_per_product: 2, + p_before: null, + }), + supabase.rpc("list_product_price_current", { p_product_ids: [input.productId] }), + ]); + if (candlesResult.error) throw candlesResult.error; + if (latestCandlesResult.error) throw latestCandlesResult.error; + if (currentResult.error) throw currentResult.error; + + const candles = groupProductPriceCandleRows( + [input.productId], + (candlesResult.data || []) as Array>, + )[input.productId] || []; + const latestCandles = groupProductPriceCandleRows( + [input.productId], + (latestCandlesResult.data || []) as Array>, + )[input.productId] || []; + const change = productPriceChange(latestCandles); + const currentRow = ((currentResult.data || []) as CurrentPriceRow[])[0]; + + return { + productId: input.productId, + interval: input.interval, + currency: "CNY", + generatedAt, + lastObservedAt: latestCandles.at(-1)?.lastObservedAt || null, + current: currentPriceFromRow(currentRow, change), + candles, + }; +} + +export async function getProductPriceChartSummaries(input: { + interval: PriceHistoryInterval; + points: 24 | 30; + platform?: string | null; + productType?: string | null; +}): Promise { + const generatedAt = new Date().toISOString(); + const products = publicCatalogProducts().filter((product) => + (!input.platform || product.platform === input.platform) && + (!input.productType || product.productType === input.productType) + ); + const productIds = products.map((product) => product.id); + const supabase = getSupabaseServerClient(); + if (!supabase || !productIds.length) { + return emptySummaryResponse(input.interval, input.points, generatedAt, productIds); + } + + const [candlesResult, currentResult] = await Promise.all([ + supabase.rpc("list_public_product_price_candles", { + p_product_ids: productIds, + p_interval: input.interval, + p_limit_per_product: input.points, + p_before: null, + }), + supabase.rpc("list_product_price_current", { p_product_ids: productIds }), + ]); + if (candlesResult.error) throw candlesResult.error; + if (currentResult.error) throw currentResult.error; + + const candlesByProduct = groupProductPriceCandleRows( + productIds, + (candlesResult.data || []) as Array>, + ); + const currentByProduct = new Map( + ((currentResult.data || []) as CurrentPriceRow[]).map((row) => [String(row.product_id || ""), row]), + ); + + return { + interval: input.interval, + points: input.points, + currency: "CNY", + generatedAt, + products: productIds.map((productId) => { + const candles = candlesByProduct[productId] || []; + const change = productPriceChange(candles); + const current = currentPriceFromRow(currentByProduct.get(productId), change); + return { + productId, + currentPrice: current.price, + eligibleOfferCount: current.eligibleOfferCount, + change: current.change, + changePercent: current.changePercent, + lastObservedAt: candles.at(-1)?.lastObservedAt || null, + candles: candles.map(compactProductPriceCandle), + }; + }), + }; +} + +export async function recordProductPriceSamples(observedAt = new Date().toISOString()) { + const supabase = getSupabaseServerClient(); + if (!supabase) throw new Error("Supabase 尚未配置。"); + const { data, error } = await supabase.rpc("record_product_price_samples", { + p_product_ids: null, + p_observed_at: observedAt, + }); + if (error) throw error; + return data; +} + +export async function pruneProductPriceHistory(input: { batchSize: number; dryRun: boolean }) { + const supabase = getSupabaseServerClient(); + if (!supabase) throw new Error("Supabase 尚未配置。"); + const { data, error } = await supabase.rpc("prune_product_price_history", { + p_batch_size: input.batchSize, + p_dry_run: input.dryRun, + }); + if (error) throw error; + return data; +} + +function currentPriceFromRow( + row: CurrentPriceRow | undefined, + change: Pick, +): ProductPriceCurrent { + const price = Number(row?.current_price); + const eligibleOfferCount = Number(row?.eligible_offer_count); + const observedAt = isoDate(row?.quoted_at); + return { + price: Number.isFinite(price) && price > 0 ? price : null, + eligibleOfferCount: Number.isInteger(eligibleOfferCount) && eligibleOfferCount > 0 ? eligibleOfferCount : 0, + observedAt, + ...change, + }; +} + +function emptyCandleResponse( + productId: string, + interval: PriceHistoryInterval, + generatedAt: string, +): ProductPriceCandleResponse { + return { + productId, + interval, + currency: "CNY", + generatedAt, + lastObservedAt: null, + current: { price: null, eligibleOfferCount: 0, observedAt: null, change: null, changePercent: null }, + candles: [], + }; +} + +function emptySummaryResponse( + interval: PriceHistoryInterval, + points: 24 | 30, + generatedAt: string, + productIds: string[], +): ProductPriceChartSummaryResponse { + return { + interval, + points, + currency: "CNY", + generatedAt, + products: productIds.map((productId) => ({ + productId, + currentPrice: null, + eligibleOfferCount: 0, + change: null, + changePercent: null, + lastObservedAt: null, + candles: [], + })), + }; +} + +function isoDate(value: unknown): string | null { + if (!value) return null; + const parsed = new Date(String(value)); + return Number.isFinite(parsed.getTime()) ? parsed.toISOString() : null; +} diff --git a/src/lib/price-history.ts b/src/lib/price-history.ts new file mode 100644 index 0000000..5eeeb3e --- /dev/null +++ b/src/lib/price-history.ts @@ -0,0 +1,174 @@ +export const PRICE_HISTORY_INTERVALS = ["1h", "1d"] as const; + +export type PriceHistoryInterval = (typeof PRICE_HISTORY_INTERVALS)[number]; + +export type ProductPriceCandle = { + time: string; + open: number; + high: number; + low: number; + close: number; + sampleCount: number; + eligibleOfferCount: number; + firstObservedAt: string; + lastObservedAt: string; +}; + +export type ProductPriceCurrent = { + price: number | null; + eligibleOfferCount: number; + observedAt: string | null; + change: number | null; + changePercent: number | null; +}; + +export type ProductPriceCandleResponse = { + productId: string; + interval: PriceHistoryInterval; + currency: "CNY"; + generatedAt: string; + lastObservedAt: string | null; + current: ProductPriceCurrent; + candles: ProductPriceCandle[]; +}; + +export type CompactProductPriceCandle = readonly [ + time: string, + open: number, + high: number, + low: number, + close: number, + sampleCount: number, +]; + +export type ProductPriceChartSummary = { + productId: string; + currentPrice: number | null; + eligibleOfferCount: number; + change: number | null; + changePercent: number | null; + lastObservedAt: string | null; + candles: CompactProductPriceCandle[]; +}; + +export type ProductPriceChartSummaryResponse = { + interval: PriceHistoryInterval; + points: 24 | 30; + currency: "CNY"; + generatedAt: string; + products: ProductPriceChartSummary[]; +}; + +export type ProductPriceChange = { + change: number | null; + changePercent: number | null; +}; + +const DEFAULT_LIMITS: Record = { "1h": 168, "1d": 90 }; +const MAX_LIMITS: Record = { "1h": 720, "1d": 365 }; + +export function parsePriceHistoryInterval(value: string | null | undefined): PriceHistoryInterval | null { + return PRICE_HISTORY_INTERVALS.find((interval) => interval === value) || null; +} + +export function parsePriceHistoryLimit( + interval: PriceHistoryInterval, + value: string | number | null | undefined, +): number | null { + if (value === null || value === undefined || value === "") return DEFAULT_LIMITS[interval]; + const parsed = typeof value === "number" ? value : Number(value); + if (!Number.isInteger(parsed) || parsed < 1 || parsed > MAX_LIMITS[interval]) return null; + return parsed; +} + +export function parsePriceChartPoints( + interval: PriceHistoryInterval, + value: string | number | null | undefined, +): 24 | 30 | null { + if (value === null || value === undefined || value === "") return interval === "1h" ? 24 : 30; + const parsed = typeof value === "number" ? value : Number(value); + return parsed === 24 || parsed === 30 ? parsed : null; +} + +export function parseExclusiveBefore(value: string | null | undefined): string | null | undefined { + if (value === null || value === undefined || value === "") return undefined; + if (!/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}(?::\d{2}(?:\.\d{1,3})?)?(?:Z|[+-]\d{2}:\d{2})$/.test(value)) { + return null; + } + const parsed = new Date(value); + return Number.isFinite(parsed.getTime()) ? parsed.toISOString() : null; +} + +export function groupProductPriceCandleRows( + productIds: string[], + rows: Array>, +): Record { + const products = Object.fromEntries(productIds.map((id) => [id, [] as ProductPriceCandle[]])); + + for (const row of rows) { + const productId = String(row.product_id || ""); + if (!Object.hasOwn(products, productId)) continue; + const candle = productPriceCandleFromRow(row); + if (candle) products[productId].push(candle); + } + + for (const candles of Object.values(products)) { + candles.sort((left, right) => left.time.localeCompare(right.time)); + } + + return products; +} + +export function productPriceCandleFromRow(row: Record): ProductPriceCandle | null { + const time = isoDate(row.period_start); + const firstObservedAt = isoDate(row.first_sample_at); + const lastObservedAt = isoDate(row.last_sample_at); + const values = [row.open_price, row.high_price, row.low_price, row.close_price].map(Number); + if (!time || !firstObservedAt || !lastObservedAt || values.some((value) => !Number.isFinite(value) || value <= 0)) { + return null; + } + + const [open, high, low, close] = values; + if (high < Math.max(open, close) || low > Math.min(open, close)) return null; + + return { + time, + open, + high, + low, + close, + sampleCount: positiveInteger(row.sample_count), + eligibleOfferCount: positiveInteger(row.eligible_offer_count), + firstObservedAt, + lastObservedAt, + }; +} + +export function productPriceChange(candles: ProductPriceCandle[]): ProductPriceChange { + const previousClose = candles.at(-2)?.close; + const latestClose = candles.at(-1)?.close; + if (previousClose === undefined || latestClose === undefined || previousClose <= 0) { + return { change: null, changePercent: null }; + } + + const change = latestClose - previousClose; + return { + change, + changePercent: (change / previousClose) * 100, + }; +} + +export function compactProductPriceCandle(candle: ProductPriceCandle): CompactProductPriceCandle { + return [candle.time, candle.open, candle.high, candle.low, candle.close, candle.sampleCount]; +} + +function isoDate(value: unknown): string | null { + if (!value) return null; + const date = new Date(String(value)); + return Number.isFinite(date.getTime()) ? date.toISOString() : null; +} + +function positiveInteger(value: unknown): number { + const parsed = Number(value); + return Number.isInteger(parsed) && parsed > 0 ? parsed : 0; +} From 04280972b21fb4152a34371e1d2782e29e6de014 Mon Sep 17 00:00:00 2001 From: xiaohe Date: Mon, 20 Jul 2026 19:12:54 +0800 Subject: [PATCH 3/4] feat(ui): render product price candles --- package-lock.json | 16 + package.json | 6 +- src/app/products/[id]/page.tsx | 3 + src/components/PriceCandleThumbnail.tsx | 201 +++++++++ src/components/PriceExplorer.tsx | 112 ++++- src/components/ProductPriceHistoryChart.tsx | 451 ++++++++++++++++++++ 6 files changed, 785 insertions(+), 4 deletions(-) create mode 100644 src/components/PriceCandleThumbnail.tsx create mode 100644 src/components/ProductPriceHistoryChart.tsx diff --git a/package-lock.json b/package-lock.json index 2266243..36347ca 100644 --- a/package-lock.json +++ b/package-lock.json @@ -12,6 +12,7 @@ "@supabase/ssr": "^0.12.0", "@supabase/supabase-js": "^2.105.3", "html-to-image": "^1.11.13", + "lightweight-charts": "^5.2.0", "lucide-react": "^1.14.0", "next": "16.2.9", "next-mdx-remote": "^6.0.0", @@ -7350,6 +7351,12 @@ "integrity": "sha512-fjquC59cD7CyW6urNXK0FBufkZcoiGG80wTuPujX590cB5Ttln20E2UB4S/WARVqhXffZl2LNgS+gQdPIIim/g==", "license": "MIT" }, + "node_modules/fancy-canvas": { + "version": "2.1.0", + "resolved": "https://registry.npmjs.org/fancy-canvas/-/fancy-canvas-2.1.0.tgz", + "integrity": "sha512-nifxXJ95JNLFR2NgRV4/MxVP45G9909wJTEKz5fg/TZS20JJZA6hfgRVh/bC9bwl2zBtBNcYPjiBE4njQHVBwQ==", + "license": "MIT" + }, "node_modules/fast-deep-equal": { "version": "3.1.3", "resolved": "https://registry.npmjs.org/fast-deep-equal/-/fast-deep-equal-3.1.3.tgz", @@ -9285,6 +9292,15 @@ "url": "https://opencollective.com/parcel" } }, + "node_modules/lightweight-charts": { + "version": "5.2.0", + "resolved": "https://registry.npmjs.org/lightweight-charts/-/lightweight-charts-5.2.0.tgz", + "integrity": "sha512-ey3Vas8UhV06ni+LT9TA1nEe4y8So4Mi6CL/oarNHFMyTktz/xy8e8+oh04Q//eO3t6etvFXgayz2fClyFQb5w==", + "license": "Apache-2.0", + "dependencies": { + "fancy-canvas": "2.1.0" + } + }, "node_modules/locate-path": { "version": "6.0.0", "resolved": "https://registry.npmjs.org/locate-path/-/locate-path-6.0.0.tgz", diff --git a/package.json b/package.json index c835ab7..96a6513 100644 --- a/package.json +++ b/package.json @@ -29,7 +29,7 @@ "start": "node_modules/node/bin/node node_modules/next/dist/bin/next start", "lint": "node_modules/node/bin/node node_modules/eslint/bin/eslint.js", "typecheck": "node_modules/node/bin/node node_modules/next/dist/bin/next typegen && node_modules/node/bin/node node_modules/typescript/bin/tsc --noEmit", - "test": "npm run test:auth-security && npm run test:feedback && npm run test:wholesale && npm run test:channel-submissions && npm run test:api-transit && npm run test:product-offer-pagination && npm run test:catalog && npm run test:api-models && npm run test:official-parser && npm run test:buyer-fee", + "test": "npm run test:auth-security && npm run test:feedback && npm run test:wholesale && npm run test:channel-submissions && npm run test:api-transit && npm run test:product-offer-pagination && npm run test:price-history && npm run test:catalog && npm run test:api-models && npm run test:official-parser && npm run test:buyer-fee", "collect:prices": "node scripts/collect-prices.mjs", "collect:shop-vip": "node scripts/collect-prices.mjs --all --kind shopApi --shop-scheduler --shop-scheduler-group vip_15m --post", "rebalance:source-shards": "node scripts/rebalance-source-shards.mjs", @@ -49,6 +49,9 @@ "test:channel-submissions": "node scripts/run-ts-test.mjs scripts/test-channel-submissions.ts", "test:auth-security": "node scripts/run-ts-test.mjs scripts/test-auth-security.ts", "test:product-offer-pagination": "node scripts/run-ts-test.mjs scripts/test-product-offer-pagination.ts && node scripts/run-ts-test.mjs scripts/test-product-offer-filters.ts", + "test:price-history": "node scripts/run-ts-test.mjs scripts/test-price-history.ts", + "test:price-history-db": "node scripts/test-price-history-db.mjs", + "test:price-history-api": "node scripts/test-price-history-api.mjs", "collect:worker": "node scripts/collect-worker.mjs", "import:api-models": "node scripts/import-api-models.mjs", "check:p2": "node scripts/p2-closeout-check.mjs", @@ -71,6 +74,7 @@ "@supabase/ssr": "^0.12.0", "@supabase/supabase-js": "^2.105.3", "html-to-image": "^1.11.13", + "lightweight-charts": "^5.2.0", "lucide-react": "^1.14.0", "next": "16.2.9", "next-mdx-remote": "^6.0.0", diff --git a/src/app/products/[id]/page.tsx b/src/app/products/[id]/page.tsx index 81e0c17..0858b02 100644 --- a/src/app/products/[id]/page.tsx +++ b/src/app/products/[id]/page.tsx @@ -7,6 +7,7 @@ import { BrandIcon } from "@/components/BrandIcon"; import { JsonLd } from "@/components/JsonLd"; import { ProductDetailHeader, ProductReturnLink } from "@/components/ProductDetailHeader"; import { ProductOffersPanel } from "@/components/ProductOffersPanel"; +import { ProductPriceHistoryChart } from "@/components/ProductPriceHistoryChart"; import { publicCatalogProducts } from "@/lib/catalog"; import { getPublicProductSummary, listPublicProductOffers } from "@/lib/data"; import { @@ -133,6 +134,8 @@ export default async function ProductDetail({ + +

渠道报价表

diff --git a/src/components/PriceCandleThumbnail.tsx b/src/components/PriceCandleThumbnail.tsx new file mode 100644 index 0000000..a5feb8c --- /dev/null +++ b/src/components/PriceCandleThumbnail.tsx @@ -0,0 +1,201 @@ +"use client"; + +import { RefreshCw } from "lucide-react"; +import { useCallback, useEffect, useMemo, useState } from "react"; +import type { + PriceHistoryInterval, + ProductPriceChartSummary, + ProductPriceChartSummaryResponse, +} from "@/lib/price-history"; + +type SummaryRequestState = { + requestKey: string; + response: ProductPriceChartSummaryResponse | null; + error: string | null; +}; + +export function useProductPriceChartSummaries(input: { + interval: PriceHistoryInterval; + platform?: string | null; + productType?: string | null; + enabled?: boolean; +}) { + const platform = normalizedFilter(input.platform); + const productType = normalizedFilter(input.productType); + const points = input.interval === "1h" ? 24 : 30; + const cacheKey = `${input.interval}:${points}:${platform || "all"}:${productType || "all"}`; + const [attempt, setAttempt] = useState(0); + const requestKey = `${cacheKey}:${attempt}`; + const [state, setState] = useState(null); + const activeState = state?.requestKey === requestKey ? state : null; + const activeResponse = activeState?.response || null; + const loading = input.enabled !== false && !activeState; + const error = activeState?.error || null; + + useEffect(() => { + if (input.enabled === false) return; + + const controller = new AbortController(); + const params = new URLSearchParams({ interval: input.interval, points: String(points) }); + if (platform) params.set("platform", platform); + if (productType) params.set("productType", productType); + + void fetch(`/api/price-chart-summaries?${params.toString()}`, { signal: controller.signal }) + .then(async (fetchResponse) => { + if (!fetchResponse.ok) { + const payload = await fetchResponse.json().catch(() => null) as { message?: string } | null; + throw new Error(payload?.message || "价格走势加载失败"); + } + return fetchResponse.json() as Promise; + }) + .then((value) => { + setState({ requestKey, response: value, error: null }); + }) + .catch((fetchError: unknown) => { + if (!controller.signal.aborted) { + setState({ + requestKey, + response: null, + error: fetchError instanceof Error ? fetchError.message : "价格走势加载失败", + }); + } + }); + + return () => controller.abort(); + }, [input.enabled, input.interval, platform, points, productType, requestKey]); + + const summaries = useMemo( + () => new Map((activeResponse?.products || []).map((summary) => [summary.productId, summary])), + [activeResponse], + ); + const retry = useCallback(() => { + setAttempt((value) => value + 1); + }, []); + + return { response: activeResponse, summaries, loading, error, retry }; +} + +export function PriceCandleThumbnail({ + summary, + interval, + loading = false, + error = null, + onRetry, + className = "", +}: { + summary?: ProductPriceChartSummary; + interval: PriceHistoryInterval; + loading?: boolean; + error?: string | null; + onRetry?: () => void; + className?: string; +}) { + const points = summary?.candles || []; + const intervalLabel = interval === "1h" ? "1小时" : "1天"; + + if (loading && !summary) return ; + if (error && !summary) { + return ( +
+ 趋势加载失败 + +
+ ); + } + if (!points.length) { + return ( +
+ 暂无价格趋势 + {intervalLabel} +
+ ); + } + + const width = 148; + const height = 42; + const padding = 3; + const low = Math.min(...points.map((candle) => candle[3])); + const high = Math.max(...points.map((candle) => candle[2])); + const range = high - low || Math.max(0.01, high * 0.04); + const step = (width - padding * 2) / points.length; + const bodyWidth = Math.max(1.5, Math.min(4, step * 0.58)); + const yForPrice = (price: number) => padding + (1 - (price - low) / range) * (height - padding * 2); + const change = summary?.change ?? null; + const changePercent = summary?.changePercent ?? null; + const changeTone = change === null || Math.abs(change) < 0.000001 + ? "text-[#7a8182]" + : change > 0 + ? "text-[#b43c34]" + : "text-[#2f7a4b]"; + + return ( +
+ + + {points.map((candle, index) => { + const x = padding + step * index + step / 2; + const openY = yForPrice(candle[1]); + const closeY = yForPrice(candle[4]); + const highY = yForPrice(candle[2]); + const lowY = yForPrice(candle[3]); + const color = candle[4] > candle[1] ? "#c6453d" : candle[4] < candle[1] ? "#2f7a4b" : "#7a8182"; + return ( + + + + + ); + })} + +
+ {points.length < 2 ? "数据积累中" : intervalLabel} + {formatChange(change, changePercent)} +
+
+ ); +} + +function ThumbnailLoading({ className }: { className: string }) { + return ( +
+
+
+
+ ); +} + +function normalizedFilter(value: string | null | undefined): string | null { + const normalized = value?.trim(); + return !normalized || normalized === "全部" ? null : normalized; +} + +function formatChange(change: number | null, percent: number | null): string { + if (change === null || percent === null) return "-"; + const sign = change > 0 ? "+" : ""; + const digits = Math.abs(change) < 0.01 && change !== 0 ? 4 : 2; + return `${sign}${change.toFixed(digits)} / ${sign}${percent.toFixed(1)}%`; +} + +function formatChangeLabel(change: number | null, percent: number | null): string { + const text = formatChange(change, percent); + return text === "-" ? "" : `，涨跌 ${text}`; +} diff --git a/src/components/PriceExplorer.tsx b/src/components/PriceExplorer.tsx index 4576b91..7d43729 100644 --- a/src/components/PriceExplorer.tsx +++ b/src/components/PriceExplorer.tsx @@ -24,6 +24,7 @@ import { CategoryTabBar, CategoryTabStrip, type CategoryTabItem } from "@/compon import { CollectorSourceLogo } from "@/components/MerchantCollectorSource"; import { GuidePromptStrip } from "@/components/GuidePromptStrip"; import { MerchantFeedbackDialog, OfferActions, OfferFeedbackButton, OfferFeedbackDialog, OfferLink } from "@/components/ProductOffersPanel"; +import { PriceCandleThumbnail, useProductPriceChartSummaries } from "@/components/PriceCandleThumbnail"; import { SiteHeader } from "@/components/SiteHeader"; import { clearFeedbackResumeRequest, getFeedbackResumeRequest } from "@/lib/feedback-draft"; import { listDetailNavigationHref, shouldHandleListDetailClick } from "@/lib/list-return"; @@ -49,6 +50,10 @@ import { import { PRICE_DATA_CACHE_TTL_MS } from "@/lib/public-cache-policy"; import { PUBLIC_MERCHANT_PAGE_SIZE } from "@/lib/public-merchant-policy"; import { PUBLIC_OFFER_DEFAULT_LIMIT } from "@/lib/public-offer-query"; +import type { + PriceHistoryInterval, + ProductPriceChartSummary, +} from "@/lib/price-history"; import type { CanonicalProduct, ExplorerData, @@ -191,6 +196,7 @@ export function PriceExplorer({ const [minPrice, setMinPrice] = useState(initialState.minPrice ?? ""); const [maxPrice, setMaxPrice] = useState(initialState.maxPrice ?? ""); const [filtersOpen, setFiltersOpen] = useState(false); + const [priceHistoryInterval, setPriceHistoryInterval] = useState("1d"); const initialScopeMode = initialState.scopeMode ?? "products"; const initialViewMode = normalizeViewModeForScope(initialScopeMode, initialState.viewMode); const [viewMode, setViewMode] = useState(initialViewMode); @@ -319,6 +325,12 @@ export function PriceExplorer({ const scopedMaxPrice = filterScope.showPrice ? maxPrice : ""; const scopedMerchantCollector = filterScope.showMerchantFilters ? merchantCollector : "all"; const scopedMerchantSignal = filterScope.showMerchantFilters ? merchantSignal : "all"; + const priceHistory = useProductPriceChartSummaries({ + interval: priceHistoryInterval, + platform, + productType: scopedProductType, + enabled: !showingOffers && !showingMerchants, + }); const title = buildTitle(platform, scopedProductType, scopeMode); const searchPlaceholder = searchPlaceholderForScope(scopeMode); const activeFilterChips = buildActiveFilterChips({ @@ -1100,6 +1112,15 @@ export function PriceExplorer({
) : null} + {!showingOffers && !showingMerchants ? ( +
+ +
+ ) : null} + {showingMerchants ? ( merchantsLoading && !visibleMerchantResponse ? ( @@ -1190,13 +1211,30 @@ export function PriceExplorer({ {renderMobileProductList ? (
{products.map((product) => ( - + ))}
) : null} {renderDesktopProductTable ? (
- +
) : null} @@ -1230,20 +1268,31 @@ export function PriceExplorer({ function ProductTable({ products, returnQuery, + priceSummaries, + priceHistoryInterval, + priceHistoryLoading, + priceHistoryError, + onPriceHistoryRetry, }: { products: ExplorerProductSummary[]; returnQuery: string; + priceSummaries: Map; + priceHistoryInterval: PriceHistoryInterval; + priceHistoryLoading: boolean; + priceHistoryError: string | null; + onPriceHistoryRetry: () => void; }) { return (
- +
标准商品平台类型最低价 + 价格走势质保最低价库存渠道 @@ -1300,6 +1349,15 @@ function ProductTable({ +
+ + void; }) { const previewOffer = product.lowestOffer; const available = product.inStockCount > 0; @@ -2115,6 +2183,16 @@ function MobileProductCard({ +
+ +
+
有货 {product.inStockCount} 缺货 {product.outOfStockCount} @@ -2125,6 +2203,34 @@ function MobileProductCard({ ); } +function PriceHistoryIntervalSwitch({ + interval, + onChange, +}: { + interval: PriceHistoryInterval; + onChange: (value: PriceHistoryInterval) => void; +}) { + return ( +
+ {(["1h", "1d"] as const).map((value) => ( + + ))} +
+ ); +} + function WarrantyLowestPrice({ product, returnQuery, diff --git a/src/components/ProductPriceHistoryChart.tsx b/src/components/ProductPriceHistoryChart.tsx new file mode 100644 index 0000000..c710d8a --- /dev/null +++ b/src/components/ProductPriceHistoryChart.tsx @@ -0,0 +1,451 @@ +"use client"; + +import { RefreshCw, RotateCcw } from "lucide-react"; +import { useCallback, useEffect, useMemo, useRef, useState } from "react"; +import type { + CandlestickData, + IChartApi, + ISeriesApi, + MouseEventParams, + Time, + UTCTimestamp, +} from "lightweight-charts"; +import type { + PriceHistoryInterval, + ProductPriceCandle, + ProductPriceCandleResponse, +} from "@/lib/price-history"; +import { formatCurrency, formatRelativeTime } from "@/lib/utils"; + +const INTERVAL_LIMITS: Record = { "1h": 168, "1d": 90 }; +const EMPTY_CANDLES: ProductPriceCandle[] = []; + +type CandleRequestState = { + requestKey: string; + response: ProductPriceCandleResponse | null; + error: string | null; +}; + +export function ProductPriceHistoryChart({ + productId, + productName, +}: { + productId: string; + productName: string; +}) { + const [interval, setInterval] = useState("1d"); + const { response, loading, error, retry } = useProductPriceCandles(productId, interval); + const candles = response?.candles || EMPTY_CANDLES; + const [activeCandle, setActiveCandle] = useState(null); + const [resetVersion, setResetVersion] = useState(0); + const displayCandle = activeCandle || candles.at(-1) || null; + const displayChange = useMemo(() => candleChange(candles, displayCandle), [candles, displayCandle]); + + return ( +
+
+
+

+ 最低可买价走势 +

+

+ {productName} · 最低起购量为 1 或无批量限制 · 每根 K 线代表 {interval === "1h" ? "1 小时" : "1 天"} +

+
+
+ + { + setActiveCandle(null); + setInterval(value); + }} + /> +
+
+ + + +
+ {loading && !response ? ( + + ) : error && !response ? ( + + {error} + + + ) : candles.length ? ( + <> + + + + ) : ( + + {response?.current.price ? "首批价格已记录，K 线数据积累中" : "暂无历史数据"} + + )} +
+ +
+ + + + + +
+ +

+ K 线基于符合条件的最低公开报价生成，仅供价格观察，不代表实际成交价格。 +

+
+ ); +} + +function useProductPriceCandles(productId: string, interval: PriceHistoryInterval) { + const cacheKey = `${productId}:${interval}`; + const [attempt, setAttempt] = useState(0); + const requestKey = `${cacheKey}:${attempt}`; + const [state, setState] = useState(null); + const activeState = state?.requestKey === requestKey ? state : null; + const activeResponse = activeState?.response || null; + const loading = !activeState; + const error = activeState?.error || null; + + useEffect(() => { + const controller = new AbortController(); + const params = new URLSearchParams({ interval, limit: String(INTERVAL_LIMITS[interval]) }); + void fetch(`/api/products/${encodeURIComponent(productId)}/price-candles?${params.toString()}`, { + signal: controller.signal, + }) + .then(async (fetchResponse) => { + if (!fetchResponse.ok) { + const payload = await fetchResponse.json().catch(() => null) as { message?: string } | null; + throw new Error(payload?.message || "价格走势加载失败"); + } + return fetchResponse.json() as Promise; + }) + .then((value) => { + setState({ requestKey, response: value, error: null }); + }) + .catch((fetchError: unknown) => { + if (!controller.signal.aborted) { + setState({ + requestKey, + response: null, + error: fetchError instanceof Error ? fetchError.message : "价格走势加载失败", + }); + } + }); + + return () => controller.abort(); + }, [interval, productId, requestKey]); + + const retry = useCallback(() => { + setAttempt((value) => value + 1); + }, []); + + return { response: activeResponse, loading, error, retry }; +} + +function CurrentPriceStrip({ + response, + loading, +}: { + response: ProductPriceCandleResponse | null; + loading: boolean; +}) { + const current = response?.current; + const priceText = current?.price ? formatCurrency(current.price, "CNY") : "当前暂无有效报价"; + return ( +
+
+

当前最低可买价

+

{loading && !response ? "加载中" : priceText}

+
+ + + +
+ ); +} + +function CandlestickCanvas({ + candles, + interval, + resetVersion, + onActiveCandleChange, +}: { + candles: ProductPriceCandle[]; + interval: PriceHistoryInterval; + resetVersion: number; + onActiveCandleChange: (candle: ProductPriceCandle | null) => void; +}) { + const containerRef = useRef(null); + const chartRef = useRef(null); + + useEffect(() => { + const container = containerRef.current; + if (!container) return; + let disposed = false; + let chart: IChartApi | null = null; + let resizeObserver: ResizeObserver | null = null; + let unsubscribe: (() => void) | null = null; + + void import("lightweight-charts").then(({ CandlestickSeries, ColorType, CrosshairMode, createChart }) => { + if (disposed) return; + chart = createChart(container, { + width: container.clientWidth, + height: container.clientHeight, + layout: { + background: { type: ColorType.Solid, color: "#ffffff" }, + textColor: "#687071", + attributionLogo: true, + fontFamily: "var(--font-sans), ui-sans-serif, system-ui, sans-serif", + fontSize: 11, + }, + crosshair: { mode: CrosshairMode.Normal }, + grid: { vertLines: { color: "#f0f2f2" }, horzLines: { color: "#edf0f1" } }, + rightPriceScale: { borderColor: "#dfe4e5", scaleMargins: { top: 0.18, bottom: 0.12 } }, + timeScale: { + borderColor: "#dfe4e5", + timeVisible: interval === "1h", + secondsVisible: false, + rightOffset: 2, + barSpacing: interval === "1h" ? 7 : 9, + minBarSpacing: 3, + tickMarkFormatter: (time: Time) => formatChartTime(time, interval, true), + }, + localization: { + locale: "zh-CN", + priceFormatter: (price: number) => formatCompactPrice(price), + timeFormatter: (time: Time) => formatChartTime(time, interval, false), + }, + handleScroll: true, + handleScale: true, + }); + chartRef.current = chart; + const precision = pricePrecision(candles); + const series: ISeriesApi<"Candlestick"> = chart.addSeries(CandlestickSeries, { + upColor: "#c6453d", + downColor: "#2f7a4b", + wickUpColor: "#c6453d", + wickDownColor: "#2f7a4b", + borderVisible: false, + priceFormat: { type: "price", precision, minMove: 10 ** -precision }, + }); + const candleByTimestamp = new Map(); + const chartData: CandlestickData[] = candles.map((candle) => { + const timestamp = Math.floor(new Date(candle.time).getTime() / 1000) as UTCTimestamp; + candleByTimestamp.set(timestamp, candle); + return { time: timestamp, open: candle.open, high: candle.high, low: candle.low, close: candle.close }; + }); + series.setData(chartData); + chart.timeScale().fitContent(); + + const handleCrosshairMove = (param: MouseEventParams) => { + onActiveCandleChange(typeof param.time === "number" ? candleByTimestamp.get(param.time) || null : null); + }; + chart.subscribeCrosshairMove(handleCrosshairMove); + unsubscribe = () => chart?.unsubscribeCrosshairMove(handleCrosshairMove); + + resizeObserver = new ResizeObserver((entries) => { + const rect = entries[0]?.contentRect; + if (rect?.width && rect.height) chart?.applyOptions({ width: rect.width, height: rect.height }); + }); + resizeObserver.observe(container); + }); + + return () => { + disposed = true; + resizeObserver?.disconnect(); + unsubscribe?.(); + chart?.remove(); + if (chartRef.current === chart) chartRef.current = null; + }; + }, [candles, interval, onActiveCandleChange]); + + useEffect(() => { + if (resetVersion > 0) chartRef.current?.timeScale().fitContent(); + }, [resetVersion]); + + return
; +} + +function CandleTooltip({ + candle, + change, + interval, +}: { + candle: ProductPriceCandle | null; + change: { change: number | null; changePercent: number | null }; + interval: PriceHistoryInterval; +}) { + if (!candle) return null; + return ( +
+

{formatCandleTime(candle.time, interval)}

+

+ 开 {formatCompactPrice(candle.open)} · 高 {formatCompactPrice(candle.high)} · 低 {formatCompactPrice(candle.low)} · 收 {formatCompactPrice(candle.close)} +

+

样本 {candle.sampleCount} · 较前期 {formatSignedChange(change.change, change.changePercent)}

+
+ ); +} + +function IntervalSwitch({ + interval, + onChange, +}: { + interval: PriceHistoryInterval; + onChange: (value: PriceHistoryInterval) => void; +}) { + return ( +
+ {(["1h", "1d"] as const).map((value) => ( + + ))} +
+ ); +} + +function PriceFact({ + label, + value, + tone = "default", +}: { + label: string; + value: string; + tone?: "default" | "up" | "down" | "muted"; +}) { + const valueClass = { default: "text-[#202829]", up: "text-[#b43c34]", down: "text-[#2f7a4b]", muted: "text-[#7a8182]" }[tone]; + return ( +
+
{label}
+
{value}
+
+ ); +} + +function ChartLoading() { + return ( +
+ {Array.from({ length: 22 }, (_, index) => ( +
+ ))} +
+ ); +} + +function ChartMessage({ children }: { children: React.ReactNode }) { + return ( +
+ {children} +
+ ); +} + +function candleChange(candles: ProductPriceCandle[], candle: ProductPriceCandle | null) { + if (!candle) return { change: null, changePercent: null }; + const index = candles.findIndex((item) => item.time === candle.time); + const previousClose = index > 0 ? candles[index - 1]?.close : undefined; + if (!previousClose) return { change: null, changePercent: null }; + const change = candle.close - previousClose; + return { change, changePercent: (change / previousClose) * 100 }; +} + +function changeTone(value: number | null): "up" | "down" | "muted" { + if (value === null || Math.abs(value) < 0.000001) return "muted"; + return value > 0 ? "up" : "down"; +} + +function formatSignedChange(change: number | null, percent: number | null): string { + if (change === null || percent === null) return "-"; + const sign = change > 0 ? "+" : ""; + const direction = change > 0 ? "上涨" : change < 0 ? "下跌" : "持平"; + return `${direction} ${sign}${formatNumber(change)} (${sign}${percent.toFixed(2)}%)`; +} + +function formatCompactPrice(value: number): string { + return `¥${formatNumber(value)}`; +} + +function formatNumber(value: number): string { + const digits = Math.abs(value) < 0.01 && value !== 0 ? 4 : 2; + return new Intl.NumberFormat("zh-CN", { maximumFractionDigits: digits }).format(value); +} + +function pricePrecision(candles: ProductPriceCandle[]): number { + const minimum = Math.min(...candles.flatMap((candle) => [candle.open, candle.high, candle.low, candle.close])); + return minimum < 0.01 ? 4 : 2; +} + +function formatCandleTime(value: string, interval: PriceHistoryInterval): string { + return new Intl.DateTimeFormat("zh-CN", { + timeZone: "Asia/Shanghai", + year: "numeric", + month: "2-digit", + day: "2-digit", + ...(interval === "1h" ? { hour: "2-digit", minute: "2-digit", hour12: false } : {}), + }).format(new Date(value)); +} + +function formatChartTime(time: Time, interval: PriceHistoryInterval, compact: boolean): string { + if (typeof time !== "number") { + if (typeof time === "string") return time; + return `${time.year}-${String(time.month).padStart(2, "0")}-${String(time.day).padStart(2, "0")}`; + } + const options: Intl.DateTimeFormatOptions = interval === "1h" + ? { timeZone: "Asia/Shanghai", month: compact ? undefined : "2-digit", day: compact ? undefined : "2-digit", hour: "2-digit", minute: "2-digit", hour12: false } + : { timeZone: "Asia/Shanghai", year: compact ? undefined : "numeric", month: "2-digit", day: "2-digit" }; + return new Intl.DateTimeFormat("zh-CN", options).format(new Date(time * 1000)); +} From c63c9c494750d0d92231cf0d7fc5a4f61a6283c7 Mon Sep 17 00:00:00 2001 From: xiaohe Date: Mon, 20 Jul 2026 19:14:11 +0800 Subject: [PATCH 4/4] test: cover product price history --- scripts/test-price-history-api.mjs | 107 +++++++++ scripts/test-price-history-db.mjs | 360 +++++++++++++++++++++++++++++ scripts/test-price-history.ts | 96 ++++++++ 3 files changed, 563 insertions(+) create mode 100644 scripts/test-price-history-api.mjs create mode 100644 scripts/test-price-history-db.mjs create mode 100644 scripts/test-price-history.ts diff --git a/scripts/test-price-history-api.mjs b/scripts/test-price-history-api.mjs new file mode 100644 index 0000000..f26f5fc --- /dev/null +++ b/scripts/test-price-history-api.mjs @@ -0,0 +1,107 @@ +#!/usr/bin/env node + +import { spawn } from "node:child_process"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; + +const repoRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), ".."); +const port = 32000 + (process.pid % 10000); +const baseUrl = `http://127.0.0.1:${port}`; +const node = path.join(repoRoot, "node_modules", "node", "bin", "node"); +const next = path.join(repoRoot, "node_modules", "next", "dist", "bin", "next"); +const log = []; +const server = spawn(node, [next, "dev", "--webpack", "--hostname", "127.0.0.1", "--port", String(port)], { + cwd: repoRoot, + env: { + ...process.env, + NEXT_PUBLIC_SUPABASE_URL: "", + SUPABASE_SERVICE_ROLE_KEY: "", + }, + stdio: ["ignore", "pipe", "pipe"], +}); + +server.stdout.on("data", (chunk) => rememberLog(chunk)); +server.stderr.on("data", (chunk) => rememberLog(chunk)); + +try { + await waitForServer(); + + const oneHour = await getJson("/api/products/chatgpt-plus/price-candles?interval=1h&limit=720", 200); + assert(oneHour.body.productId === "chatgpt-plus", "detail API did not normalize the product ID"); + assert(oneHour.body.interval === "1h", "detail API did not return 1h"); + assert(Array.isArray(oneHour.body.candles) && oneHour.body.candles.length === 0, "unconfigured detail API did not return empty history"); + assert(oneHour.body.current?.price === null, "unconfigured detail API returned a synthetic current price"); + assertCacheHeaders(oneHour.response); + + const oneDay = await getJson("/api/products/chatgpt-plus/price-candles?interval=1d&limit=365&before=2026-07-20T00%3A00%3A00Z", 200); + assert(oneDay.body.interval === "1d", "detail API did not return 1d"); + assertCacheHeaders(oneDay.response); + + await getJson("/api/products/chatgpt-plus/price-candles?interval=4h", 400); + await getJson("/api/products/chatgpt-plus/price-candles?interval=1h&limit=721", 400); + await getJson("/api/products/chatgpt-plus/price-candles?interval=1d&before=2026-07-20", 400); + await getJson("/api/products/not-a-product/price-candles?interval=1d", 404); + + const summaries1h = await getJson("/api/price-chart-summaries?interval=1h&points=24&platform=ChatGPT", 200); + assert(summaries1h.body.interval === "1h" && summaries1h.body.points === 24, "summary API did not return 1h/24"); + assert(Array.isArray(summaries1h.body.products) && summaries1h.body.products.length > 1, "summary API did not return a product batch"); + assert(summaries1h.body.products.every((item) => item.candles.length === 0), "unconfigured summary API returned synthetic candles"); + assertCacheHeaders(summaries1h.response); + + const summaries1d = await getJson("/api/price-chart-summaries?interval=1d&points=30&productType=%E8%AE%A2%E9%98%85%2F%E4%BC%9A%E5%91%98", 200); + assert(summaries1d.body.interval === "1d" && summaries1d.body.points === 30, "summary API did not return 1d/30"); + assert(!containsInternalField(summaries1d.body), "public price history API exposed internal quote fields"); + await getJson("/api/price-chart-summaries?interval=1d&points=25", 400); + + console.log("product price history API test passed"); +} catch (error) { + const suffix = log.length ? `\nNext dev output:\n${log.join("")}` : ""; + throw new Error(`${error instanceof Error ? error.message : String(error)}${suffix}`); +} finally { + server.kill("SIGTERM"); + await Promise.race([ + new Promise((resolve) => server.once("exit", resolve)), + new Promise((resolve) => setTimeout(resolve, 5000)), + ]); + if (server.exitCode === null) server.kill("SIGKILL"); +} + +async function waitForServer() { + for (let attempt = 0; attempt < 120; attempt += 1) { + if (server.exitCode !== null) throw new Error(`Next dev exited with ${server.exitCode}`); + try { + const response = await fetch(`${baseUrl}/api/price-chart-summaries?interval=1d&points=30`); + if (response.status === 200) return; + } catch { + // The dev server is still compiling. + } + await new Promise((resolve) => setTimeout(resolve, 500)); + } + throw new Error("Next dev did not become ready within 60 seconds"); +} + +async function getJson(pathname, expectedStatus) { + const response = await fetch(`${baseUrl}${pathname}`); + const body = await response.json().catch(() => null); + assert(response.status === expectedStatus, `${pathname} expected ${expectedStatus}, got ${response.status}: ${JSON.stringify(body)}`); + return { response, body }; +} + +function assertCacheHeaders(response) { + assert(response.headers.get("cdn-cache-control") === "public, s-maxage=300", "CDN cache TTL is not 300 seconds"); + assert(response.headers.get("cloudflare-cdn-cache-control") === "public, s-maxage=300", "Cloudflare cache TTL is not 300 seconds"); +} + +function containsInternalField(value) { + const text = JSON.stringify(value); + return /offer_id|source_id|confirmation|consecutive_valid/i.test(text); +} + +function rememberLog(chunk) { + log.push(String(chunk)); + while (log.join("").length > 12000) log.shift(); +} + +function assert(condition, message) { + if (!condition) throw new Error(message); +} diff --git a/scripts/test-price-history-db.mjs b/scripts/test-price-history-db.mjs new file mode 100644 index 0000000..322dc06 --- /dev/null +++ b/scripts/test-price-history-db.mjs @@ -0,0 +1,360 @@ +#!/usr/bin/env node + +import crypto from "node:crypto"; +import { readFileSync } from "node:fs"; +import { spawnSync } from "node:child_process"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; + +const repoRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), ".."); +const schema = readFileSync(path.join(repoRoot, "supabase", "schema.sql"), "utf8"); +const docker = commandPath("docker"); +assert(docker, "未找到 docker，无法执行价格历史 PostgreSQL 集成测试。"); + +const image = process.env.PRICEAI_TEST_POSTGRES_IMAGE || "postgres:18-alpine"; +const container = `priceai-price-history-${process.pid}-${Date.now()}`; +const password = crypto.randomBytes(18).toString("hex"); + +try { + run(docker, ["run", "--rm", "--detach", "--name", container, "--env", `POSTGRES_PASSWORD=${password}`, image]); + waitForPostgres(docker, container); + runPsql(` + create role anon nologin; + create role authenticated nologin; + create role service_role nologin; + create schema auth; + create function auth.uid() returns uuid language sql stable as $$ select null::uuid $$; + `); + runPsql(schema); + runPsql(testSql()); + console.log("product price history database test passed"); +} finally { + spawnSync(docker, ["rm", "--force", container], { stdio: "ignore" }); +} + +function runPsql(sql) { + const result = spawnSync( + docker, + ["exec", "-i", container, "psql", "-X", "-v", "ON_ERROR_STOP=1", "-U", "postgres", "-d", "postgres"], + { input: sql, encoding: "utf8", maxBuffer: 32 * 1024 * 1024, stdio: ["pipe", "ignore", "pipe"] }, + ); + if (result.status !== 0) throw new Error(result.stderr || "价格历史 PostgreSQL 集成测试失败。"); +} + +function waitForPostgres(command, name) { + let consecutiveReady = 0; + for (let attempt = 0; attempt < 60; attempt += 1) { + const probe = spawnSync(command, ["exec", name, "pg_isready", "-U", "postgres"], { stdio: "ignore" }); + consecutiveReady = probe.status === 0 ? consecutiveReady + 1 : 0; + if (consecutiveReady >= 2) return; + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 500); + } + throw new Error("隔离 PostgreSQL 在 30 秒内没有就绪。"); +} + +function run(command, args) { + const result = spawnSync(command, args, { encoding: "utf8" }); + if (result.status !== 0) throw new Error(result.stderr || `${command} 执行失败。`); +} + +function commandPath(name) { + const result = spawnSync("which", [name], { encoding: "utf8" }); + return result.status === 0 ? result.stdout.trim() : null; +} + +function assert(condition, message) { + if (!condition) throw new Error(message); +} + +function testSql() { + return String.raw` +set timezone = 'UTC'; + +insert into canonical_products (id, slug, display_name, platform, product_type, spec, summary) +values + ('p1', 'p1', 'Product One', 'ChatGPT', '订阅/会员', '', ''), + ('p2', 'p2', 'Product Two', 'Claude', '订阅/会员', '', ''), + ('p3', 'p3', 'Product Three', 'Gemini', '成品账号', '', ''); + +insert into sources (id, name, entry_url, enabled) +values ('s1', 'Source One', 'https://example.com', true); + +insert into raw_offers ( + id, source_id, source_name, source_title, price, currency, status, source_status, + effective_status, freshness_status, url, tags, stock_count, min_order_quantity, + hidden, canonical_product_id, captured_at, last_seen_at, verified_at, expires_at +) +values + ('valid-low', 's1', 'Source One', '独享账号', 10, 'CNY', 'in_stock', 'in_stock', 'available', 'fresh', 'https://example.com/low', '{}', 9, 1, false, 'p1', now(), now(), now(), now() + interval '7 days'), + ('valid-high', 's1', 'Source One', '独享账号', 12, 'CNY', 'low_stock', 'low_stock', 'available', 'fresh', 'https://example.com/high', '{}', null, null, false, 'p1', now(), now(), now(), now() + interval '7 days'), + ('valid-p2', 's1', 'Source One', '独享账号', 33, 'CNY', 'in_stock', 'in_stock', 'available', 'fresh', 'https://example.com/p2', '{}', 2, 1, false, 'p2', now(), now(), now(), now() + interval '7 days'), + ('invalid-hidden', 's1', 'Source One', '独享账号', 1, 'CNY', 'in_stock', 'in_stock', 'available', 'fresh', 'https://example.com/hidden', '{}', 1, 1, true, 'p1', now(), now(), now(), now() + interval '7 days'), + ('invalid-zero', 's1', 'Source One', '独享账号', 0, 'CNY', 'in_stock', 'in_stock', 'available', 'fresh', 'https://example.com/zero', '{}', 1, 1, false, 'p1', now(), now(), now(), now() + interval '7 days'), + ('invalid-currency', 's1', 'Source One', '独享账号', 1, 'USD', 'in_stock', 'in_stock', 'available', 'fresh', 'https://example.com/usd', '{}', 1, 1, false, 'p1', now(), now(), now(), now() + interval '7 days'), + ('invalid-url', 's1', 'Source One', '独享账号', 1, 'CNY', 'in_stock', 'in_stock', 'available', 'fresh', '', '{}', 1, 1, false, 'p1', now(), now(), now(), now() + interval '7 days'), + ('invalid-status', 's1', 'Source One', '独享账号', 1, 'CNY', 'out_of_stock', 'out_of_stock', 'unavailable', 'fresh', 'https://example.com/status', '{}', 1, 1, false, 'p1', now(), now(), now(), now() + interval '7 days'), + ('invalid-stock', 's1', 'Source One', '独享账号', 1, 'CNY', 'in_stock', 'in_stock', 'available', 'fresh', 'https://example.com/stock', '{}', 0, 1, false, 'p1', now(), now(), now(), now() + interval '7 days'), + ('invalid-minimum', 's1', 'Source One', '独享账号', 1, 'CNY', 'in_stock', 'in_stock', 'available', 'fresh', 'https://example.com/minimum', '{}', 1, 2, false, 'p1', now(), now(), now(), now() + interval '7 days'), + ('invalid-shared', 's1', 'Source One', '多人共享拼车', 1, 'CNY', 'in_stock', 'in_stock', 'available', 'fresh', 'https://example.com/shared', '{}', 1, 1, false, 'p1', now(), now(), now(), now() + interval '7 days'), + ('invalid-web', 's1', 'Source One', '仅限网页号', 1, 'CNY', 'in_stock', 'in_stock', 'available', 'fresh', 'https://example.com/web', '{}', 1, 1, false, 'p1', now(), now(), now(), now() + interval '7 days'), + ('invalid-mirror', 's1', 'Source One', '国内镜像站', 1, 'CNY', 'in_stock', 'in_stock', 'available', 'fresh', 'https://example.com/mirror', '{}', 1, 1, false, 'p1', now(), now(), now(), now() + interval '7 days'); + +create function test_confirm(p_offer_id text, p_confirmed_at timestamptz) +returns void +language sql +as $$ + insert into raw_offer_confirmations ( + raw_offer_id, source_id, canonical_product_id, confirmed_at, captured_at, + last_seen_at, verified_at, expires_at, source_status, effective_status, + freshness_status, source_priority, confidence, price, stock_count, updated_at + ) + select + offers.id, offers.source_id, offers.canonical_product_id, p_confirmed_at, + p_confirmed_at, p_confirmed_at, p_confirmed_at, now() + interval '7 days', + offers.source_status, offers.effective_status, offers.freshness_status, + offers.source_priority, offers.confidence, offers.price, offers.stock_count, + p_confirmed_at + from raw_offers offers + where offers.id = p_offer_id + on conflict (raw_offer_id) do update set + source_id = excluded.source_id, + canonical_product_id = excluded.canonical_product_id, + confirmed_at = excluded.confirmed_at, + captured_at = excluded.captured_at, + last_seen_at = excluded.last_seen_at, + verified_at = excluded.verified_at, + expires_at = excluded.expires_at, + source_status = excluded.source_status, + effective_status = excluded.effective_status, + freshness_status = excluded.freshness_status, + source_priority = excluded.source_priority, + confidence = excluded.confidence, + price = excluded.price, + stock_count = excluded.stock_count, + updated_at = excluded.updated_at; +$$; + +select test_confirm(id, now()) from raw_offers; + +do $$ +begin + if (select consecutive_valid_confirmations from raw_offer_confirmations where raw_offer_id = 'valid-low') <> 1 then + raise exception 'first valid confirmation was not recorded as one'; + end if; + if exists (select 1 from product_price_eligible_offers) then + raise exception 'an offer became eligible before its second confirmation'; + end if; +end; +$$; + +select test_confirm(id, now() + interval '1 minute') from raw_offers; + +update raw_offer_confirmations +set + confirmed_at = confirmed_at - interval '1 hour', + effective_status = 'unavailable' +where raw_offer_id = 'valid-low'; + +do $$ +begin + if (select count(*) from product_price_eligible_offers) <> 3 then + raise exception 'eligible offer filters or two-confirmation rule failed'; + end if; + if exists ( + select 1 from raw_offer_confirmations + where raw_offer_id like 'invalid-%' and consecutive_valid_confirmations <> 0 + ) then + raise exception 'invalid offers retained a confirmation streak'; + end if; + if (select consecutive_valid_confirmations from raw_offer_confirmations where raw_offer_id = 'valid-low') <> 2 then + raise exception 'an out-of-order confirmation overwrote the latest streak'; + end if; +end; +$$; + +update raw_offers +set source_status = 'out_of_stock' +where id = 'valid-high'; + +do $$ +begin + if (select consecutive_valid_confirmations from raw_offer_confirmations where raw_offer_id = 'valid-high') <> 0 then + raise exception 'source-status-only unavailability did not reset the confirmation streak'; + end if; +end; +$$; + +update raw_offers +set source_status = 'low_stock' +where id = 'valid-high'; +select test_confirm('valid-high', now() + interval '2 minutes'); + +do $$ +begin + if (select consecutive_valid_confirmations from raw_offer_confirmations where raw_offer_id = 'valid-high') <> 1 then + raise exception 'a restored offer skipped its first confirmation'; + end if; +end; +$$; + +select test_confirm('valid-high', now() + interval '3 minutes'); + +do $$ +begin + if (select consecutive_valid_confirmations from raw_offer_confirmations where raw_offer_id = 'valid-high') <> 2 then + raise exception 'a restored offer did not become eligible after two confirmations'; + end if; +end; +$$; + +select record_product_price_samples(null, '2026-07-20T15:05:00Z'); + +do $$ +begin + if (select price from product_price_samples where product_id = 'p1') <> 10 then + raise exception 'lowest eligible price was not sampled'; + end if; + if (select price from product_price_samples where product_id = 'p2') <> 33 then + raise exception 'product isolation failed'; + end if; + if exists (select 1 from product_price_samples where product_id = 'p3') then + raise exception 'product without an eligible price produced a sample'; + end if; +end; +$$; + +update raw_offers set price = 11 where id = 'valid-low'; +select test_confirm('valid-low', now() + interval '2 minutes'); +select test_confirm('valid-low', now() + interval '3 minutes'); +select record_product_price_samples(array['p1'], '2026-07-20T15:20:00Z'); + +update raw_offers set price = 9 where id = 'valid-low'; +select test_confirm('valid-low', now() + interval '4 minutes'); +select test_confirm('valid-low', now() + interval '5 minutes'); +select record_product_price_samples(array['p1'], '2026-07-20T15:35:00Z'); + +update raw_offers set price = 12 where id = 'valid-low'; +select test_confirm('valid-low', now() + interval '6 minutes'); +select test_confirm('valid-low', now() + interval '7 minutes'); +select record_product_price_samples(array['p1'], '2026-07-20T15:50:00Z'); + +update raw_offers set price = 8 where id = 'valid-low'; +select test_confirm('valid-low', now() + interval '8 minutes'); +select test_confirm('valid-low', now() + interval '9 minutes'); +select record_product_price_samples(array['p1'], '2026-07-20T15:48:00Z'); +select record_product_price_samples(array['p1'], '2026-07-20T15:50:00Z'); + +do $$ +declare + candle product_price_candles%rowtype; +begin + select * into candle + from product_price_candles + where product_id = 'p1' and candle_interval = '1h' and bucket_start = '2026-07-20T15:00:00Z'; + if candle.open_price <> 10 or candle.high_price <> 12 or candle.low_price <> 9 or candle.close_price <> 12 or candle.sample_count <> 4 then + raise exception 'hourly OHLC, ordering, or idempotency failed: %', row_to_json(candle); + end if; + if (select price from product_price_samples where product_id = 'p1' and bucket_start = '2026-07-20T15:45:00Z') <> 12 then + raise exception 'out-of-order sampling overwrote a newer sample'; + end if; + if not exists ( + select 1 from product_price_candles + where product_id = 'p1' and candle_interval = '1d' and bucket_start = '2026-07-19T16:00:00Z' + ) then + raise exception 'Asia/Shanghai daily bucket was not created'; + end if; +end; +$$; + +select record_product_price_samples(array['p1'], '2026-07-20T16:05:00Z'); +select record_product_price_samples(array['p1'], '2026-07-20T18:05:00Z'); + +do $$ +begin + if (select count(*) from product_price_candles where product_id = 'p1' and candle_interval = '1d') <> 2 then + raise exception 'Beijing day boundary did not create two daily candles'; + end if; + if exists ( + select 1 from product_price_candles + where product_id = 'p1' and candle_interval = '1h' and bucket_start = '2026-07-20T17:00:00Z' + ) then + raise exception 'a missing observation interval was filled artificially'; + end if; + if exists ( + select 1 from product_price_candles + where product_id = 'p1' and candle_interval = '1h' + and sample_count = 1 and (open_price <> high_price or high_price <> low_price or low_price <> close_price) + ) then + raise exception 'single-sample candle was not flat'; + end if; +end; +$$; + +update raw_offers set effective_status = 'unavailable' where id in ('valid-low', 'valid-high'); + +do $$ +begin + if exists (select 1 from list_product_price_current(array['p1'])) then + raise exception 'current quote remained after all eligible offers disappeared'; + end if; + if not exists (select 1 from list_public_product_price_candles(array['p1'], '1h', 24, null)) then + raise exception 'historical candles disappeared with the current quote'; + end if; +end; +$$; + +insert into product_price_samples ( + product_id, bucket_start, observed_at, price, currency, eligible_offer_count +) +values + ('p2', '2024-01-01T00:00:00Z', '2024-01-01T00:05:00Z', 20, 'CNY', 1), + ('p3', '2024-02-01T00:00:00Z', '2024-02-01T00:05:00Z', 30, 'CNY', 1); + +insert into product_price_candles ( + product_id, candle_interval, bucket_start, bucket_end, open_price, high_price, + low_price, close_price, currency, sample_count, eligible_offer_count, + first_sample_at, last_sample_at +) +values + ('p2', '1h', '2024-01-01T00:00:00Z', '2024-01-01T01:00:00Z', 20, 20, 20, 20, 'CNY', 1, 1, '2024-01-01T00:05:00Z', '2024-01-01T00:05:00Z'), + ('p2', '1d', '2023-12-31T16:00:00Z', '2024-01-01T16:00:00Z', 20, 20, 20, 20, 'CNY', 1, 1, '2024-01-01T00:05:00Z', '2024-01-01T00:05:00Z'); + +do $$ +declare + result jsonb; +begin + result := prune_product_price_history(100, true); + if (result ->> 'sampleCandidates')::integer <> 1 then + raise exception 'dry-run did not enforce aggregate coverage: %', result; + end if; + if not exists (select 1 from product_price_samples where product_id = 'p2' and bucket_start = '2024-01-01T00:00:00Z') then + raise exception 'dry-run deleted a sample'; + end if; + + result := prune_product_price_history(100, false); + if (result ->> 'deletedSamples')::integer <> 1 then + raise exception 'bounded retention did not delete the covered sample: %', result; + end if; + if not exists (select 1 from product_price_samples where product_id = 'p3' and bucket_start = '2024-02-01T00:00:00Z') then + raise exception 'retention deleted a sample without hourly and daily coverage'; + end if; + if not exists (select 1 from product_price_candles where product_id = 'p2' and candle_interval = '1d') then + raise exception 'daily retention was applied unexpectedly'; + end if; +end; +$$; + +do $$ +begin + if has_table_privilege('anon', 'product_price_samples', 'select') + or has_table_privilege('authenticated', 'product_price_candles', 'select') + then + raise exception 'internal price history tables are publicly readable'; + end if; + if has_function_privilege('anon', 'record_product_price_samples(text[],timestamptz)', 'execute') then + raise exception 'sampling RPC is publicly executable'; + end if; +end; +$$; +`; +} diff --git a/scripts/test-price-history.ts b/scripts/test-price-history.ts new file mode 100644 index 0000000..8e25384 --- /dev/null +++ b/scripts/test-price-history.ts @@ -0,0 +1,96 @@ +import { + compactProductPriceCandle, + groupProductPriceCandleRows, + parseExclusiveBefore, + parsePriceChartPoints, + parsePriceHistoryInterval, + parsePriceHistoryLimit, + productPriceChange, +} from "../src/lib/price-history.js"; + +assertEqual(parsePriceHistoryInterval("1h"), "1h"); +assertEqual(parsePriceHistoryInterval("1d"), "1d"); +assertEqual(parsePriceHistoryInterval("4h"), null); +assertEqual(parsePriceHistoryLimit("1h", null), 168); +assertEqual(parsePriceHistoryLimit("1d", undefined), 90); +assertEqual(parsePriceHistoryLimit("1h", "720"), 720); +assertEqual(parsePriceHistoryLimit("1h", "721"), null); +assertEqual(parsePriceHistoryLimit("1d", "0"), null); +assertEqual(parsePriceHistoryLimit("1d", "1.5"), null); +assertEqual(parsePriceChartPoints("1h", null), 24); +assertEqual(parsePriceChartPoints("1d", null), 30); +assertEqual(parsePriceChartPoints("1d", "24"), 24); +assertEqual(parsePriceChartPoints("1d", "25"), null); +assertEqual(parseExclusiveBefore("2026-07-20T12:34:56+08:00"), "2026-07-20T04:34:56.000Z"); +assertEqual(parseExclusiveBefore("2026-07-20"), null); + +const grouped = groupProductPriceCandleRows(["a", "b"], [ + row("a", "2026-07-20T00:00:00.000Z", 10, 12, 9, 11, 2), + row("a", "2026-07-20T01:00:00.000Z", 11, 13, 10, 12, 3), + row("a", "2026-07-20T02:00:00.000Z", 12, 14, 11, 13, 4), + { + ...row("b", "invalid", 1, 2, 0.5, 1.5, 1), + period_start: "invalid", + }, +]); + +assertEqual(grouped.a.length, 3); +assertEqual(grouped.a[0]?.time, "2026-07-20T00:00:00.000Z"); +assertEqual(grouped.b.length, 0); +assertDeepEqual(productPriceChange([]), { change: null, changePercent: null }); +assertDeepEqual(productPriceChange(grouped.a.slice(0, 1)), { change: null, changePercent: null }); +assertEqual(productPriceChange(grouped.a).change, 1); +assertApprox(productPriceChange(grouped.a).changePercent, 100 / 12); +assertDeepEqual(compactProductPriceCandle(grouped.a[2]!), [ + "2026-07-20T02:00:00.000Z", + 12, + 14, + 11, + 13, + 4, +]); + +console.log("price history test passed"); + +function row( + productId: string, + time: string, + open: number, + high: number, + low: number, + close: number, + sampleCount: number, +) { + return { + product_id: productId, + period_start: time, + open_price: open, + high_price: high, + low_price: low, + close_price: close, + sample_count: sampleCount, + eligible_offer_count: 2, + first_sample_at: time, + last_sample_at: time, + }; +} + +function assertEqual(actual: unknown, expected: unknown) { + if (actual !== expected) { + throw new Error(`Expected ${JSON.stringify(actual)} to equal ${JSON.stringify(expected)}.`); + } +} + +function assertDeepEqual(actual: unknown, expected: unknown) { + const actualText = JSON.stringify(actual); + const expectedText = JSON.stringify(expected); + if (actualText !== expectedText) { + throw new Error(`Expected ${actualText} to equal ${expectedText}.`); + } +} + +function assertApprox(actual: number | null, expected: number) { + if (actual === null || Math.abs(actual - expected) > 0.000001) { + throw new Error(`Expected ${JSON.stringify(actual)} to approximately equal ${expected}.`); + } +}