标准商品
平台
类型
最低价
+ 价格走势
质保最低价
库存
渠道
@@ -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}
+
+
+
+
+
+ {formatCandleTime(candle.time, interval)}
+
+ 开 {formatCompactPrice(candle.open)} · 高 {formatCompactPrice(candle.high)} · 低 {formatCompactPrice(candle.low)} · 收 {formatCompactPrice(candle.close)}
+
+ 样本 {candle.sampleCount} · 较前期 {formatSignedChange(change.change, change.changePercent)}
+
+ {(["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 (
+
+ );
+}
+
+function ChartLoading() {
+ return (
+
+ {Array.from({ length: 22 }, (_, index) => (
+
+ ))}
+
+ {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));
+}
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 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;
+}
diff --git a/supabase/migrations/20260720220000_product_price_history.sql b/supabase/migrations/20260720220000_product_price_history.sql
new file mode 100644
index 0000000..5b6f526
--- /dev/null
+++ b/supabase/migrations/20260720220000_product_price_history.sql
@@ -0,0 +1,684 @@
+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;
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;
|