diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0b4f6967..d67dd412 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -88,6 +88,9 @@ jobs: - name: Replay the DOS-Me review AI points proposal (Docker) run: node scripts/test-review-ai-points-proposal.mjs + - name: Replay the DOS-Me extension observations proposal (Docker) + run: pnpm --filter @repo/shared-data build && node scripts/test-extension-observations-proposal.mjs + - name: Check generated rate schema types match the migrations run: pnpm check:db-types diff --git a/.gitignore b/.gitignore index be909f56..3e5f451e 100644 --- a/.gitignore +++ b/.gitignore @@ -164,6 +164,9 @@ playwright-report-item-model/ !Docs/Proposals/2026-11-03-dos-me-review-ai-points.sql !Docs/Proposals/2026-11-03-dos-me-review-ai-points-rollback.sql !Docs/Proposals/2026-11-03-dos-me-review-ai-points-verify.sql +!Docs/Proposals/2026-11-04-dos-me-extension-observations.sql +!Docs/Proposals/2026-11-04-dos-me-extension-observations-rollback.sql +!Docs/Proposals/2026-11-04-dos-me-extension-observations-verify.sql * - Copy* * - Bản sao* *.dump diff --git a/Docs/Proposals/2026-11-04-dos-me-extension-observations-rollback.sql b/Docs/Proposals/2026-11-04-dos-me-extension-observations-rollback.sql new file mode 100644 index 00000000..57820b2c --- /dev/null +++ b/Docs/Proposals/2026-11-04-dos-me-extension-observations-rollback.sql @@ -0,0 +1,22 @@ +-- ROLLBACK of the PROPOSAL 2026-11-04-dos-me-extension-observations.sql (DOS/Rate#1606). +-- One transaction: drops the four functions, then rate.extension_observations, +-- rate.extension_submitters and rate.extension_ingest_secret with their policies and indexes. +-- Every quarantined observation and submitter id is lost, and so is the stored ingest key hash +-- (set it again after a re-apply). Turn RATE_EXTENSION_OBSERVATIONS off on the web first; the +-- routes would otherwise answer errors. Nothing else reads these tables. +-- Safe to run twice. + +BEGIN; + +SET LOCAL lock_timeout = '5s'; + +DROP FUNCTION IF EXISTS rate.purge_extension_observations(integer, integer, integer); +DROP FUNCTION IF EXISTS rate.delete_extension_submitter(uuid); +DROP FUNCTION IF EXISTS rate.submit_extension_observation(text, uuid, jsonb, text); +DROP FUNCTION IF EXISTS rate.register_extension_submitter(text, text); + +DROP TABLE IF EXISTS rate.extension_observations; +DROP TABLE IF EXISTS rate.extension_submitters; +DROP TABLE IF EXISTS rate.extension_ingest_secret; + +COMMIT; diff --git a/Docs/Proposals/2026-11-04-dos-me-extension-observations-verify.sql b/Docs/Proposals/2026-11-04-dos-me-extension-observations-verify.sql new file mode 100644 index 00000000..ee762c57 --- /dev/null +++ b/Docs/Proposals/2026-11-04-dos-me-extension-observations-verify.sql @@ -0,0 +1,263 @@ +-- VERIFY the PROPOSAL 2026-11-04-dos-me-extension-observations.sql (DOS/Rate#1606). Read-only. +-- Run after the migration and after the ingest key row is set, on preview, beta and production, +-- as postgres or service_role. V1 to V12 return 0. V13 is information: rows the purge should +-- already have handled. + +-- V1 the three tables exist with row level security on and not forced (the functions' owner +-- must pass it); expect 0 +SELECT 3 - count(*) FROM pg_class +WHERE oid IN (to_regclass('rate.extension_ingest_secret'), to_regclass('rate.extension_submitters'), + to_regclass('rate.extension_observations')) + AND relrowsecurity AND NOT relforcerowsecurity; + +-- V2 anon and authenticated hold no privilege on the three tables or the identity sequence, +-- and PUBLIC holds none either; expect 0 +SELECT count(*) FROM ( + SELECT r.role, t.tbl, p.priv + FROM unnest(ARRAY['anon', 'authenticated']) AS r(role) + CROSS JOIN unnest(ARRAY['rate.extension_ingest_secret', 'rate.extension_submitters', + 'rate.extension_observations']) AS t(tbl) + CROSS JOIN unnest(ARRAY['SELECT', 'INSERT', 'UPDATE', 'DELETE', 'TRUNCATE', 'REFERENCES', + 'TRIGGER']) AS p(priv) + WHERE has_table_privilege(r.role, t.tbl, p.priv) + OR has_any_column_privilege(r.role, t.tbl, 'SELECT') + OR has_any_column_privilege(r.role, t.tbl, 'INSERT') + OR has_any_column_privilege(r.role, t.tbl, 'UPDATE') + UNION ALL + SELECT r.role, 'sequence', p.priv + FROM unnest(ARRAY['anon', 'authenticated']) AS r(role) + CROSS JOIN unnest(ARRAY['USAGE', 'SELECT', 'UPDATE']) AS p(priv) + WHERE has_sequence_privilege(r.role, + pg_get_serial_sequence('rate.extension_observations', 'id'), p.priv) + UNION ALL + SELECT 'PUBLIC', c.relname::text, a.privilege_type + FROM pg_class c + CROSS JOIN LATERAL aclexplode(c.relacl) a + WHERE c.oid IN (to_regclass('rate.extension_ingest_secret'), + to_regclass('rate.extension_submitters'), + to_regclass('rate.extension_observations')) + AND a.grantee = 0 +) x; + +-- V3 service_role has full access; rate_worker reads submitters and observations, updates only +-- extension_observations.status, cannot insert, delete or truncate, and has nothing on the +-- ingest key; expect 0 +SELECT count(*) FROM ( + SELECT t.tbl, p.priv FROM unnest(ARRAY['rate.extension_ingest_secret', + 'rate.extension_submitters', + 'rate.extension_observations']) AS t(tbl) + CROSS JOIN unnest(ARRAY['SELECT', 'INSERT', 'UPDATE', 'DELETE']) AS p(priv) + WHERE NOT has_table_privilege('service_role', t.tbl, p.priv) + UNION ALL + SELECT t.tbl, 'rate_worker select' FROM unnest(ARRAY['rate.extension_submitters', + 'rate.extension_observations']) AS t(tbl) + WHERE NOT has_table_privilege('rate_worker', t.tbl, 'SELECT') + OR has_table_privilege('rate_worker', t.tbl, 'INSERT') + OR has_table_privilege('rate_worker', t.tbl, 'UPDATE') + OR has_table_privilege('rate_worker', t.tbl, 'DELETE') + OR has_table_privilege('rate_worker', t.tbl, 'TRUNCATE') + OR has_any_column_privilege('rate_worker', t.tbl, 'INSERT') + UNION ALL + SELECT 'rate.extension_submitters', 'rate_worker column update' + WHERE has_any_column_privilege('rate_worker', 'rate.extension_submitters', 'UPDATE') + UNION ALL + SELECT 'rate.extension_ingest_secret', 'rate_worker' + WHERE has_any_column_privilege('rate_worker', 'rate.extension_ingest_secret', 'SELECT') + OR has_any_column_privilege('rate_worker', 'rate.extension_ingest_secret', 'INSERT') + OR has_any_column_privilege('rate_worker', 'rate.extension_ingest_secret', 'UPDATE') + OR has_table_privilege('rate_worker', 'rate.extension_ingest_secret', 'DELETE') + OR has_table_privilege('rate_worker', 'rate.extension_ingest_secret', 'TRUNCATE') + UNION ALL + SELECT 'rate.extension_observations', a.attname::text + FROM pg_attribute a + WHERE a.attrelid = to_regclass('rate.extension_observations') AND a.attnum > 0 + AND NOT a.attisdropped + AND has_column_privilege('rate_worker', a.attrelid, a.attnum, 'UPDATE') <> (a.attname = 'status') +) x; + +-- V4 the six policies exist, and none applies to PUBLIC, anon or authenticated; expect 0 +SELECT (6 - ( + SELECT count(*) FROM pg_policies + WHERE schemaname = 'rate' AND (tablename, policyname, cmd, roles) IN ( + ('extension_ingest_secret', 'Service role access extension_ingest_secret', 'ALL', + '{service_role}'::name[]), + ('extension_submitters', 'Service role access extension_submitters', 'ALL', + '{service_role}'::name[]), + ('extension_submitters', 'Rate worker read extension_submitters', 'SELECT', + '{rate_worker}'::name[]), + ('extension_observations', 'Service role access extension_observations', 'ALL', + '{service_role}'::name[]), + ('extension_observations', 'Rate worker read extension_observations', 'SELECT', + '{rate_worker}'::name[]), + ('extension_observations', 'Rate worker status extension_observations', 'UPDATE', + '{rate_worker}'::name[])) +)) + ( + SELECT count(*) FROM pg_policies + WHERE schemaname = 'rate' + AND tablename IN ('extension_ingest_secret', 'extension_submitters', 'extension_observations') + AND roles && '{public,anon,authenticated}'::name[] +); + +-- V5 the four functions are SECURITY DEFINER with an empty search_path, and owned by the +-- tables' owner (who passes their row level security); expect 0 +SELECT 4 - count(*) FROM pg_proc p +WHERE p.oid IN ( + to_regprocedure('rate.register_extension_submitter(text, text)'), + to_regprocedure('rate.submit_extension_observation(text, uuid, jsonb, text)'), + to_regprocedure('rate.delete_extension_submitter(uuid)'), + to_regprocedure('rate.purge_extension_observations(integer, integer, integer)')) + AND p.prosecdef AND p.proconfig @> ARRAY['search_path=""'] + AND p.proowner = (SELECT relowner FROM pg_class + WHERE oid = to_regclass('rate.extension_observations')) + AND p.proowner = ALL (SELECT relowner FROM pg_class + WHERE oid IN (to_regclass('rate.extension_ingest_secret'), + to_regclass('rate.extension_submitters'))); + +-- V6 EXECUTE: PUBLIC has none; anon, authenticated and service_role call the first three; +-- only service_role calls the purge; rate_worker calls none; expect 0 +SELECT count(*) FROM ( + SELECT f.sig, a.privilege_type + FROM unnest(ARRAY['rate.register_extension_submitter(text, text)', + 'rate.submit_extension_observation(text, uuid, jsonb, text)', + 'rate.delete_extension_submitter(uuid)', + 'rate.purge_extension_observations(integer, integer, integer)']) AS f(sig) + JOIN pg_proc p ON p.oid = to_regprocedure(f.sig) + CROSS JOIN LATERAL aclexplode(COALESCE(p.proacl, acldefault('f', p.proowner))) a + WHERE a.grantee = 0 + UNION ALL + SELECT f.sig, r.role + FROM unnest(ARRAY['rate.register_extension_submitter(text, text)', + 'rate.submit_extension_observation(text, uuid, jsonb, text)', + 'rate.delete_extension_submitter(uuid)']) AS f(sig) + CROSS JOIN unnest(ARRAY['anon', 'authenticated', 'service_role']) AS r(role) + WHERE to_regprocedure(f.sig) IS NULL OR NOT has_function_privilege(r.role, f.sig, 'EXECUTE') + UNION ALL + SELECT 'purge', r.role FROM unnest(ARRAY['anon', 'authenticated']) AS r(role) + WHERE has_function_privilege(r.role, + 'rate.purge_extension_observations(integer, integer, integer)', 'EXECUTE') + UNION ALL + SELECT 'purge', 'service_role' + WHERE NOT has_function_privilege('service_role', + 'rate.purge_extension_observations(integer, integer, integer)', 'EXECUTE') + UNION ALL + SELECT f.sig, 'rate_worker' + FROM unnest(ARRAY['rate.register_extension_submitter(text, text)', + 'rate.submit_extension_observation(text, uuid, jsonb, text)', + 'rate.delete_extension_submitter(uuid)', + 'rate.purge_extension_observations(integer, integer, integer)']) AS f(sig) + WHERE has_function_privilege('rate_worker', f.sig, 'EXECUTE') +) x; + +-- V7 the CHECKs exist and are validated, and the foreign key cascades on delete; expect 0 +SELECT (20 - ( + SELECT count(*) FROM pg_constraint + WHERE contype = 'c' AND convalidated AND ( + (conrelid = to_regclass('rate.extension_observations') AND conname IN ( + 'extension_observations_schema_version_check', 'extension_observations_marketplace_check', + 'extension_observations_listing_url_check', 'extension_observations_external_item_id_check', + 'extension_observations_external_shop_id_check', 'extension_observations_sku_id_check', + 'extension_observations_variant_label_check', 'extension_observations_price_vnd_check', + 'extension_observations_list_price_vnd_check', 'extension_observations_stock_state_check', + 'extension_observations_stock_qty_check', 'extension_observations_observed_at_check', + 'extension_observations_extractor_check', 'extension_observations_network_hash_check', + 'extension_observations_status_check')) + OR (conrelid = to_regclass('rate.extension_submitters') + AND conname = 'extension_submitters_network_hash_check') + OR (conrelid = to_regclass('rate.extension_ingest_secret') + AND conname IN ('extension_ingest_secret_id_check', 'extension_ingest_secret_key_not_empty_check', + 'extension_ingest_secret_key_sha256_check'))) +)) - ( + SELECT count(*) FROM pg_constraint + WHERE conrelid = to_regclass('rate.extension_observations') AND contype = 'f' + AND confrelid = to_regclass('rate.extension_submitters') AND confdeltype = 'c' +); + +-- V8 the five indexes exist; expect 0 +SELECT 5 - count(*) FROM pg_indexes +WHERE schemaname = 'rate' AND indexname IN ( + 'extension_submitters_created_idx', 'extension_submitters_network_created_idx', + 'extension_observations_submitter_received_idx', 'extension_observations_network_received_idx', + 'extension_observations_item_received_idx'); + +-- V9 every column has the type, nullability and default of the proposal, and no other column +-- exists; expect 0 +WITH expected (tbl, col, typ, not_null, def) AS (VALUES + ('extension_ingest_secret', 'id', 'boolean', true, 'true'), + ('extension_ingest_secret', 'key_sha256', 'text', true, NULL), + ('extension_ingest_secret', 'updated_at', 'timestamp with time zone', true, 'now()'), + ('extension_submitters', 'id', 'uuid', true, 'gen_random_uuid()'), + ('extension_submitters', 'created_at', 'timestamp with time zone', true, 'now()'), + ('extension_submitters', 'network_hash', 'text', false, NULL), + ('extension_observations', 'id', 'bigint', true, 'identity always'), + ('extension_observations', 'submitter_id', 'uuid', true, NULL), + ('extension_observations', 'schema_version', 'smallint', true, NULL), + ('extension_observations', 'marketplace', 'text', true, NULL), + ('extension_observations', 'listing_url', 'text', true, NULL), + ('extension_observations', 'external_item_id', 'text', true, NULL), + ('extension_observations', 'external_shop_id', 'text', false, NULL), + ('extension_observations', 'sku_id', 'text', false, NULL), + ('extension_observations', 'variant_label', 'text', false, NULL), + ('extension_observations', 'price_vnd', 'bigint', true, NULL), + ('extension_observations', 'list_price_vnd', 'bigint', false, NULL), + ('extension_observations', 'stock_state', 'text', true, NULL), + ('extension_observations', 'stock_qty', 'integer', false, NULL), + ('extension_observations', 'observed_at', 'timestamp with time zone', true, NULL), + ('extension_observations', 'extractor', 'jsonb', true, NULL), + ('extension_observations', 'received_at', 'timestamp with time zone', true, 'now()'), + ('extension_observations', 'network_hash', 'text', false, NULL), + ('extension_observations', 'status', 'text', true, '''quarantined''::text') +), actual AS ( + SELECT c.relname::text AS tbl, a.attname::text AS col, + format_type(a.atttypid, a.atttypmod) AS typ, a.attnotnull AS not_null, + COALESCE(pg_get_expr(d.adbin, d.adrelid), + CASE a.attidentity WHEN 'a' THEN 'identity always' + WHEN 'd' THEN 'identity by default' END) AS def + FROM pg_attribute a + JOIN pg_class c ON c.oid = a.attrelid + LEFT JOIN pg_attrdef d ON d.adrelid = a.attrelid AND d.adnum = a.attnum + WHERE c.oid IN (to_regclass('rate.extension_ingest_secret'), + to_regclass('rate.extension_submitters'), + to_regclass('rate.extension_observations')) + AND a.attnum > 0 AND NOT a.attisdropped +) +SELECT count(*) FROM ( + (SELECT * FROM expected EXCEPT SELECT * FROM actual) + UNION ALL + (SELECT * FROM actual EXCEPT SELECT * FROM expected) +) x; + +-- V10 the limits are in the function bodies: 120 and 600 an hour and the 16 KB cap in submit, +-- 20 an hour and 600 a minute in register; expect 0 +SELECT count(*) FROM (VALUES + ('rate.submit_extension_observation(text, uuid, jsonb, text)', '%) >= 120 THEN%'), + ('rate.submit_extension_observation(text, uuid, jsonb, text)', '%) >= 600 THEN%'), + ('rate.submit_extension_observation(text, uuid, jsonb, text)', '%octet_length(o::text) > 16384%'), + ('rate.register_extension_submitter(text, text)', '%) >= 20 THEN%'), + ('rate.register_extension_submitter(text, text)', '%) >= 600 THEN%') +) AS t(sig, pattern) +WHERE NOT EXISTS (SELECT 1 FROM pg_proc p + WHERE p.oid = to_regprocedure(t.sig) AND p.prosrc LIKE t.pattern); + +-- V11 the ingest key table is granted to nobody but its owner and service_role; expect 0 +SELECT count(*) FROM pg_class c +CROSS JOIN LATERAL aclexplode(COALESCE(c.relacl, acldefault('r', c.relowner))) a +WHERE c.oid = to_regclass('rate.extension_ingest_secret') + AND a.grantee <> c.relowner + AND a.grantee <> (SELECT oid FROM pg_roles WHERE rolname = 'service_role'); + +-- V12 the ingest key row is set (without it, register and submit refuse every caller); expect 0 +SELECT 1 - count(*) FROM rate.extension_ingest_secret WHERE id; + +-- V13 information: rows past the default retention (non-zero means the daily purge is not +-- running) +SELECT + (SELECT count(*) FROM rate.extension_observations + WHERE network_hash IS NOT NULL AND received_at < now() - interval '29 days') + + (SELECT count(*) FROM rate.extension_submitters + WHERE network_hash IS NOT NULL AND created_at < now() - interval '29 days') AS hashes_due, + (SELECT count(*) FROM rate.extension_observations + WHERE received_at < now() - interval '181 days') AS observations_due, + (SELECT count(*) FROM rate.extension_submitters s + WHERE s.created_at < now() - interval '8 days' + AND NOT EXISTS (SELECT 1 FROM rate.extension_observations o + WHERE o.submitter_id = s.id)) AS empty_submitters_due; diff --git a/Docs/Proposals/2026-11-04-dos-me-extension-observations.md b/Docs/Proposals/2026-11-04-dos-me-extension-observations.md new file mode 100644 index 00000000..56fe41dc --- /dev/null +++ b/Docs/Proposals/2026-11-04-dos-me-extension-observations.md @@ -0,0 +1,224 @@ +# DOS-Me proposal: extension observation quarantine (M2) + +DOS/Rate#1606, milestone M2 of `Docs/Proposals/2026-10-28-extension-data-source.md`, with the +contract's Revision 2 (after four reviews). Status: **proposal, nothing applied.** It goes +through DOS-Me after JOY approves it. + +| File | What | +| ------------------------------------------------------- | --------------------------------------------------------- | +| `2026-11-04-dos-me-extension-observations.sql` | the migration, one transaction | +| `2026-11-04-dos-me-extension-observations-verify.sql` | read-only checks, V1 to V12 return 0 (V13 is information) | +| `2026-11-04-dos-me-extension-observations-rollback.sql` | drops the functions and the three tables | +| `scripts/test-extension-observations-proposal.mjs` | replays it all on PostgreSQL in Docker | + +## What it does + +The extension sends what a shopper's browser shows on a listing: price, list price, stock and +variant. This migration stores those observations in a quarantine. Nothing reads them into +prices, products or ratings. Corroboration and promotion are M3/M4, in DOS/Rate-Services. + +- **`rate.extension_ingest_secret`** (new): one private row, the sha256 of the web's secret + `RATE_EXTENSION_INGEST_KEY`. + - The migration inserts no row. DOS-Me sets it at apply time. The key itself is never in git. + - Only `service_role` has access. +- **`rate.extension_submitters`** (new): one row per install that opted in. It holds a random + id, its creation time and the network hash it registered from. +- **`rate.extension_observations`** (new): the quarantine. + - One column per key of the `ExtensionObservation v1` payload, each with a CHECK that repeats + the payload rule. + - `received_at`, set by the server. The server trusts it more than `observed_at`. + - `network_hash`: the web's HMAC-SHA256 of the sender's /24 or /48. + - `status`: `quarantined`, `corroborated` or `rejected`. + - It cascades from its submitter. +- **Access.** + - Row level security is on for the three tables, and not forced. + - `anon` and `authenticated` have no table privilege; they use the RPCs only. + - `service_role` has full access. + - `rate_worker` reads submitters and observations and may update `status` only (a column + grant). It has nothing on the ingest key. +- **RPCs.** All four are `SECURITY DEFINER` with `search_path = ''`, owned by the tables' owner. + EXECUTE is revoked from PUBLIC. + 1. `rate.register_extension_submitter(p_ingest_key, p_network_hash)`, for anon, + authenticated and service_role. + - A key that is missing, shorter than 32 characters, or whose sha256 is not the stored row + raises 42501, DETAIL `ingest_key`. Without a row, every call is refused. The row can + never hold the sha256 of the empty string (a CHECK). + - A network hash that is not 64 lowercase hex characters raises 22023, DETAIL + `network_hash`. + - Over 20 new submitters per network hash in the last hour raises 54000, DETAIL + `network_hash`. + - Over 600 new submitters in the last minute, in all, raises 54000, DETAIL `submitter`. + - Each count runs under an advisory lock, so it is exact. + 2. `rate.submit_extension_observation(...)`, for anon, authenticated and service_role. Its + arguments: `p_ingest_key`, `p_submitter_id`, `p_observation`, `p_network_hash`. + - The ingest key and the network hash are checked as in register. + - Then every payload rule, in the order of the shared validator. A refusal raises 22023, + and DETAIL names the field: + - `observation` for a payload that is not an object, is over 16 KB, or has another key. + A caller's key name is never echoed. + - then each key in turn, its presence checked with its rule; a nullable key must be + present as null; + - then `extractor.source` and `extractor.extension_version`. + - An unknown submitter raises P0002 with DETAIL `submitter_id`. + - Over 120 observations per submitter, or 600 per network hash, in the last hour, it + raises 54000. DETAIL is `submitter_id` or `network_hash`. + - It returns `{"id", "status": "quarantined", "received_at"}`. + 3. `rate.delete_extension_submitter(p_submitter_id)`, for anon, authenticated and + service_role. + - It takes no ingest key, so erasure always works. + - It deletes the submitter, and its observations by cascade. + - It returns how many observations went. An unknown id returns 0. + 4. `rate.purge_extension_observations(...)`, for service_role only. Its arguments: + `p_network_hash_days` (28), `p_keep_days` (180), `p_empty_submitter_days` (7). + - It clears `network_hash` on older submitters, then on older observations. That order + matches delete (the submitter, then its observations), so a delete during a purge waits + instead of deadlocking. + - It deletes older observations. + - It deletes submitters without observations that are older than + `p_empty_submitter_days`. + - It locks them first, skipping any a submit holds, then deletes them under a fresh + check. + - So an observation committed during the purge is never cascaded away. + - It returns the four counts. + +Marketplace ids are the platform ids of the extension and the crawl: `shopee`, `lazada`, `tiki`, +`tiktok`, `cellphones`, `tgdd`, `fptshop`, `hoangha`, `didongviet`, `phongvu`. The host table: + +| marketplace | hosts | +| ----------- | -------------------------------- | +| shopee | shopee.vn | +| lazada | lazada.vn and subdomains | +| tiki | tiki.vn | +| tiktok | www.tiktok.com, shop.tiktok.com | +| cellphones | cellphones.com.vn | +| tgdd | thegioididong.com and subdomains | +| fptshop | fptshop.com.vn | +| hoangha | hoanghamobile.com | +| didongviet | didongviet.vn | +| phongvu | phongvu.vn | + +A listing URL is https, at most 2048 characters. It has no `?`, `#` or `;`, no space and no +control character. Its host is lowercase, with no user and no port. + +## The ingest key + +- Without the key, a caller holding only the public anon key can no longer call register or + submit. So nobody can bypass the web: its flag, its validator, its salt and the network + limit. +- The network hash is therefore always the web's. It is evidence of independence only for rows + written through the web, and with the key in place that means all rows. +- The key is a server secret of the web. It must never be committed. Its sha256 may be. +- **Rotation is a hard cut.** Once the new row is set, register and submit answer 42501 to the + old key, so the web's writes fail (503) until it redeploys with the new key. Deletes keep + working. + +## Privacy + +- **The submitter id is pseudonymous personal data.** It links every observation of one + install. It holds no account, email or device data. +- **The network hash is pseudonymous personal data too.** It is keyed by a web secret, so the + database alone cannot reverse it. It is kept for the limits and for abuse review. +- **Retention:** + - `network_hash` is cleared after 28 days, on observations and on submitters; + - observations are deleted after 180 days; + - submitters without observations are deleted after 7 days. + - All three are done by `rate.purge_extension_observations`. It still needs a daily job as + `service_role`: the DOS-Me pg_cron or an RS worker. V13 of the verify file shows rows it + should already have handled. +- **Delete on request:** the extension calls the web's `DELETE` route, which calls + `rate.delete_extension_submitter`. Everything that install sent goes at once. +- **No other personal data:** the payload refuses any other key. It has no ratings, account + data, cookies, query strings or referrers. + +## Order and dependencies + +- **Needs:** the `rate` schema, the API roles and `rate_worker` + (`20261018183000_rate_worker_role.sql`). All are in production. +- **Shared validator:** the replay compares the RPC with `parseExtensionObservationV1` in + `packages/shared-data/src/rules/extension-observation.ts`. That code PR merges first. +- **Web:** the routes behind `RATE_EXTENSION_OBSERVATIONS=1` (off by default) call RPCs 1 to 3 + with the ingest key. Turn the flag on only after the apply, the key row and the verify. +- **After this:** M3/M4 in DOS/Rate-Services read the quarantine as `rate_worker` and set + `status`. + +**Rating impact: none.** Nothing reads these tables into prices, products or ratings. + +## Decisions and limits + +1. **The limits are coarse.** + - The register brake of 600 a minute is global. A flood of registrations from many networks + blocks new installs for a minute, but not installs that already have an id. + - Per submitter, per network and in all, calls run one at a time (a row lock and advisory + locks), so the counts are exact. +2. **The submitter id is the only key to its data.** Whoever holds an id can delete it. The id + is random (122 bits) and stays in the extension's storage. +3. **The CHECKs repeat the payload rules** so that `service_role`'s direct writes keep to them. + One difference: a CHECK cannot read the clock, so `observed_at` is held to `received_at` + (10 minutes ahead, 24 hours behind), not to `now()`. +4. **A change to the payload needs a migration.** The SQL repeats the rules of the shared + validator, which is plain functions. A drift fails closed, and the replay compares both + over 167 payloads. + - `observed_at` takes seconds 00 to 59 (no leap second) and an offset up to 15:59, which is + what Postgres parses. +5. **Where the RPC and the validator differ on purpose:** + - the 16 KB cap is the RPC's alone (the web caps the request body); + - a lone UTF-16 surrogate: the validator names the field, and jsonb refuses the text + itself (22P02). +6. **Two indexes are not in the contract:** `extension_submitters_created_idx` and + `extension_submitters_network_created_idx`, for the register limits and the purge. + +## Apply + +These steps are for the session or agent that applies DOS-Me. JOY runs no command. + +1. **Generate the key** with a CSPRNG, for example `openssl rand -hex 32`. Keep it out of git, + logs and chat. +2. **Hash the exact bytes** with `printf %s "$KEY" | sha256sum`. + - Never `echo`, which adds a newline. + - Never in SQL, where the key would reach the database logs. + - Check that `$KEY` is set: the sha256 of an empty key is refused by a CHECK. +3. **Set the web's secrets** on every Vercel environment that points at that database: + - `RATE_EXTENSION_INGEST_KEY`: the key; + - `RATE_NETWORK_HASH_SALT`: random, at least 16 characters. + - Preview, beta and production share the prod database, so they share one key. +4. **Apply the migration.** Copy the `.sql` into DOS-Me `supabase/migrations/` with a new + timestamp, and the verify and rollback files into `scripts/ops/`. +5. **Insert the row**, as `postgres` or `service_role`, with only the hash: + + ```sql + INSERT INTO rate.extension_ingest_secret (key_sha256) VALUES ('') + ON CONFLICT (id) DO UPDATE SET key_sha256 = EXCLUDED.key_sha256, updated_at = now(); + ``` + +6. **Run the verify file** read-only. V1 to V12 must return 0; V12 proves the row is set. +7. **Only then turn the web flag on** (`RATE_EXTENSION_OBSERVATIONS=1`). +8. Schedule the daily purge as `service_role`. + +**Rollback:** the rollback file, safe to run twice. Turn the web flag off first. Every +observation and submitter id is lost, and the key row with them. + +## Tests (local, Docker) + +`pnpm --filter @repo/shared-data build && node scripts/test-extension-observations-proposal.mjs`: + +- the proposal applied twice, and V1 to V12; V1, V2 and V11 fail on a forced RLS or a stray + grant; +- the ingest key: no row, a wrong, empty or null key, and the hash itself are refused (42501); + a 31-character key is refused even when its hash is stored; the empty key's hash cannot be + stored; the right key passes; the table holds one row; +- a required network hash in register and submit; +- 39 valid and 128 refused payloads, each with the same verdict and field from the RPC and from + `parseExtensionObservationV1`, URL and time edges included; +- an unknown or null submitter (P0002), and nothing written by a refused call; +- the limits at their exact thresholds: the 120th per submitter, the 600th per network (with + 100 older rows), the 20th register per network and the 600th register in a minute pass; the + next one is refused; +- delete: the count, the cascade, 0 for an unknown or null id, no key needed; +- purge: hashes cleared on both tables, rows and empty submitters deleted, service_role only; + a submit held open during the purge keeps its submitter and its row; and a delete in the + middle of a stalled purge waits for it instead of deadlocking (the old lock order fails this + test with "deadlock detected"); +- grants: anon and authenticated reach none of the three tables; rate_worker reads and updates + `status` only and has nothing on the key; the CHECKs hold `service_role`; +- the rollback twice, then the proposal again (refusing until the key row is set), and again + over data. diff --git a/Docs/Proposals/2026-11-04-dos-me-extension-observations.sql b/Docs/Proposals/2026-11-04-dos-me-extension-observations.sql new file mode 100644 index 00000000..7a214a38 --- /dev/null +++ b/Docs/Proposals/2026-11-04-dos-me-extension-observations.sql @@ -0,0 +1,659 @@ +-- ============================================================================= +-- PROPOSAL (DOS/Rate#1606): milestone M2 of Docs/Proposals/2026-10-28-extension-data-source.md, +-- the quarantine for ExtensionObservation v1. A DOS-Me migration to review and apply; nothing +-- here is applied by DOS/Rate. Verify: 2026-11-04-dos-me-extension-observations-verify.sql +-- (V1 to V12 return 0). Rollback: 2026-11-04-dos-me-extension-observations-rollback.sql. +-- Contract: the M2 contract with its Revision 2 (ingest key, required network hash, register +-- limits, purge race fix). +-- +-- What it does: +-- rate.extension_ingest_secret one private row: the sha256 of the web's ingest key. This file +-- inserts no row; DOS-Me sets it at apply time. Without it, or +-- with a wrong key, register and submit refuse (42501). +-- rate.extension_submitters one row per extension install that opted in: a random id, its +-- creation time and the network hash it registered from. +-- Pseudonymous personal data. +-- rate.extension_observations the quarantine: one row per observation the extension sent, +-- one column per payload key, checked like the payload, plus +-- received_at, a salted network hash and a status +-- (quarantined, corroborated or rejected). +-- All three tables: row level security on; no anon or authenticated access; service_role +-- full; rate_worker reads submitters and observations and may update status only. +-- rate.register_extension_submitter(key, h) anon with the ingest key: a new submitter +-- id; 20 an hour per network hash, 600 a minute in all (54000). +-- rate.submit_extension_observation(key, id, obs, h) anon with the ingest key: one +-- observation, validated field by field (22023 with the field in +-- DETAIL), 120 an hour per submitter and 600 an hour per network +-- hash (54000). +-- rate.delete_extension_submitter(id) anon, no key: deletes the submitter and its +-- observations; returns how many observations went (0 for an +-- unknown id). +-- rate.purge_extension_observations(d, k, e) service_role: clears network hashes after d +-- days (28), deletes observations after k days (180) and +-- submitters without observations after e days (7). +-- +-- Nothing here writes price_history, products or any public table. Promotion out of the +-- quarantine is M3/M4, in DOS/Rate-Services. +-- +-- Depends on: the rate schema, the API roles (anon, authenticated, service_role) and rate_worker +-- (20261018183000_rate_worker_role.sql); all are in production. +-- +-- Rating impact: none. No rating, price or product reads these tables. +-- +-- Idempotent: IF NOT EXISTS for the tables and indexes, CREATE OR REPLACE for functions, +-- DROP POLICY IF EXISTS before each policy, grants restated every run. +-- ============================================================================= + +BEGIN; + +-- New tables only; the FK takes a lock on its own new parent. Give up rather than queue. +SET LOCAL lock_timeout = '5s'; + +-- 1. The ingest key ---------------------------------------------------------------------------- +-- One row at most (id is always true). Only the key's sha256 is stored; the key is the web's +-- secret RATE_EXTENSION_INGEST_KEY and is never in git. DOS-Me inserts the row at apply time. + +CREATE TABLE IF NOT EXISTS rate.extension_ingest_secret ( + id boolean PRIMARY KEY DEFAULT true CHECK (id), + key_sha256 text NOT NULL CHECK (key_sha256 ~ '^[0-9a-f]{64}$'), + updated_at timestamptz NOT NULL DEFAULT now(), + -- Not the sha256 of the empty string: a runbook that hashed an unset variable would + -- otherwise store a key that an empty caller matches. + CONSTRAINT extension_ingest_secret_key_not_empty_check + CHECK (key_sha256 <> 'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855') +); + +COMMENT ON TABLE rate.extension_ingest_secret IS + 'sha256 of the web''s extension ingest key (DOS/Rate#1606). register_extension_submitter and submit_extension_observation refuse a caller without the key. service_role only.'; + +-- 2. Submitters -------------------------------------------------------------------------------- + +CREATE TABLE IF NOT EXISTS rate.extension_submitters ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + created_at timestamptz NOT NULL DEFAULT now(), + network_hash text, + CONSTRAINT extension_submitters_network_hash_check CHECK (network_hash ~ '^[0-9a-f]{64}$') +); + +COMMENT ON TABLE rate.extension_submitters IS + 'One row per extension install that opted in to sending observations (DOS/Rate#1606). Pseudonymous personal data: delete on request through rate.delete_extension_submitter; network_hash cleared after 28 days; deleted 7 days after creation when it has no observations, else with its last observation.'; + +-- The global register brake and the purge read created_at; the per-network register limit +-- reads network_hash. +CREATE INDEX IF NOT EXISTS extension_submitters_created_idx + ON rate.extension_submitters (created_at); +CREATE INDEX IF NOT EXISTS extension_submitters_network_created_idx + ON rate.extension_submitters (network_hash, created_at) WHERE network_hash IS NOT NULL; + +-- 3. The quarantine ---------------------------------------------------------------------------- +-- Each CHECK repeats a payload rule of rate.submit_extension_observation, which refuses first +-- with the field's name; the CHECKs hold service_role's direct writes to the same rules. + +CREATE TABLE IF NOT EXISTS rate.extension_observations ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + submitter_id uuid NOT NULL REFERENCES rate.extension_submitters(id) ON DELETE CASCADE, + schema_version smallint NOT NULL, + marketplace text NOT NULL, + listing_url text NOT NULL, + external_item_id text NOT NULL, + external_shop_id text, + sku_id text, + variant_label text, + price_vnd bigint NOT NULL, + list_price_vnd bigint, + stock_state text NOT NULL, + stock_qty integer, + observed_at timestamptz NOT NULL, + extractor jsonb NOT NULL, + received_at timestamptz NOT NULL DEFAULT now(), + network_hash text, + status text NOT NULL DEFAULT 'quarantined', + CONSTRAINT extension_observations_schema_version_check CHECK (schema_version = 1), + CONSTRAINT extension_observations_marketplace_check CHECK (marketplace IN ( + 'shopee', 'lazada', 'tiki', 'tiktok', 'cellphones', 'tgdd', 'fptshop', + 'hoangha', 'didongviet', 'phongvu')), + -- https, at most 2048 characters, no query, fragment or path parameter, no space (JavaScript's + -- \s set) or control character, and a lowercase host without user or port that belongs to the + -- marketplace (the host table). + CONSTRAINT extension_observations_listing_url_check CHECK ( + char_length(listing_url) <= 2048 + AND listing_url !~ '[?#;[:space:]\x01-\x1f\x7f-\x9f\xa0\x1680\x2000-\x200a\x2028\x2029\x202f\x205f\x3000\xfeff]' + AND CASE marketplace + WHEN 'shopee' THEN listing_url ~ '^https://shopee\.vn(/|$)' + WHEN 'lazada' THEN listing_url ~ '^https://([a-z0-9-]+\.)*lazada\.vn(/|$)' + WHEN 'tiki' THEN listing_url ~ '^https://tiki\.vn(/|$)' + WHEN 'tiktok' THEN listing_url ~ '^https://(www|shop)\.tiktok\.com(/|$)' + WHEN 'cellphones' THEN listing_url ~ '^https://cellphones\.com\.vn(/|$)' + WHEN 'tgdd' THEN listing_url ~ '^https://([a-z0-9-]+\.)*thegioididong\.com(/|$)' + WHEN 'fptshop' THEN listing_url ~ '^https://fptshop\.com\.vn(/|$)' + WHEN 'hoangha' THEN listing_url ~ '^https://hoanghamobile\.com(/|$)' + WHEN 'didongviet' THEN listing_url ~ '^https://didongviet\.vn(/|$)' + WHEN 'phongvu' THEN listing_url ~ '^https://phongvu\.vn(/|$)' + ELSE false END), + CONSTRAINT extension_observations_external_item_id_check + CHECK (external_item_id ~ '^[A-Za-z0-9_.-]{1,64}$'), + CONSTRAINT extension_observations_external_shop_id_check + CHECK (external_shop_id ~ '^[A-Za-z0-9_.-]{1,64}$'), + CONSTRAINT extension_observations_sku_id_check CHECK (sku_id ~ '^[A-Za-z0-9_.-]{1,64}$'), + CONSTRAINT extension_observations_variant_label_check CHECK ( + char_length(variant_label) BETWEEN 1 AND 120 AND variant_label !~ '[\x01-\x1f\x7f-\x9f]'), + CONSTRAINT extension_observations_price_vnd_check + CHECK (price_vnd BETWEEN 1000 AND 2000000000), + CONSTRAINT extension_observations_list_price_vnd_check + CHECK (list_price_vnd >= price_vnd AND list_price_vnd <= 2000000000), + CONSTRAINT extension_observations_stock_state_check + CHECK (stock_state IN ('in_stock', 'low_stock', 'out_of_stock', 'unknown')), + CONSTRAINT extension_observations_stock_qty_check CHECK (stock_qty BETWEEN 0 AND 1000000), + -- The RPC checks observed_at against now(); a CHECK can only hold it to received_at. + CONSTRAINT extension_observations_observed_at_check CHECK ( + observed_at <= received_at + interval '10 minutes' + AND observed_at >= received_at - interval '24 hours'), + CONSTRAINT extension_observations_extractor_check CHECK ( + jsonb_typeof(extractor) = 'object' + AND extractor - 'source' - 'extension_version' = '{}'::jsonb + AND extractor ->> 'source' IN ('json_ld', 'dom', 'mixed') + AND jsonb_typeof(extractor -> 'source') = 'string' + AND jsonb_typeof(extractor -> 'extension_version') = 'string' + AND (extractor ->> 'extension_version') ~ '^[0-9A-Za-z.+-]{1,20}$'), + CONSTRAINT extension_observations_network_hash_check CHECK (network_hash ~ '^[0-9a-f]{64}$'), + CONSTRAINT extension_observations_status_check + CHECK (status IN ('quarantined', 'corroborated', 'rejected')) +); + +COMMENT ON TABLE rate.extension_observations IS + 'Quarantine of ExtensionObservation v1 (DOS/Rate#1606). Unverified crowd data: nothing reads it into prices or ratings until M3/M4 corroborates it. network_hash is cleared after 28 days and rows are deleted after 180 days by rate.purge_extension_observations.'; +COMMENT ON COLUMN rate.extension_observations.network_hash IS + 'HMAC-SHA256 of the sender''s /24 (IPv4) or /48 (IPv6), keyed by a web secret; pseudonymous personal data, cleared after 28 days.'; + +CREATE INDEX IF NOT EXISTS extension_observations_submitter_received_idx + ON rate.extension_observations (submitter_id, received_at); +CREATE INDEX IF NOT EXISTS extension_observations_network_received_idx + ON rate.extension_observations (network_hash, received_at) WHERE network_hash IS NOT NULL; +CREATE INDEX IF NOT EXISTS extension_observations_item_received_idx + ON rate.extension_observations (marketplace, external_item_id, received_at); + +-- 4. Grants and row level security ------------------------------------------------------------- +-- anon and authenticated reach the tables only through the RPCs below. + +ALTER TABLE rate.extension_ingest_secret ENABLE ROW LEVEL SECURITY; +ALTER TABLE rate.extension_submitters ENABLE ROW LEVEL SECURITY; +ALTER TABLE rate.extension_observations ENABLE ROW LEVEL SECURITY; + +-- The ingest key: service_role only (the functions read it as their owner). +REVOKE ALL ON rate.extension_ingest_secret FROM PUBLIC, anon, authenticated, rate_worker; +GRANT ALL ON rate.extension_ingest_secret TO service_role; +DROP POLICY IF EXISTS "Service role access extension_ingest_secret" ON rate.extension_ingest_secret; +CREATE POLICY "Service role access extension_ingest_secret" ON rate.extension_ingest_secret + FOR ALL TO service_role USING (true) WITH CHECK (true); + +REVOKE ALL ON rate.extension_submitters, rate.extension_observations + FROM PUBLIC, anon, authenticated, rate_worker; +GRANT ALL ON rate.extension_submitters, rate.extension_observations TO service_role; +GRANT SELECT ON rate.extension_submitters, rate.extension_observations TO rate_worker; +GRANT UPDATE (status) ON rate.extension_observations TO rate_worker; + +-- The identity sequence: only service_role (a project's default privileges may grant it wider). +DO $$ +DECLARE + v_seq text := pg_get_serial_sequence('rate.extension_observations', 'id'); +BEGIN + EXECUTE format('REVOKE ALL ON SEQUENCE %s FROM PUBLIC, anon, authenticated, rate_worker', v_seq); + EXECUTE format('GRANT ALL ON SEQUENCE %s TO service_role', v_seq); +END $$; + +DROP POLICY IF EXISTS "Service role access extension_submitters" ON rate.extension_submitters; +CREATE POLICY "Service role access extension_submitters" ON rate.extension_submitters + FOR ALL TO service_role USING (true) WITH CHECK (true); +DROP POLICY IF EXISTS "Rate worker read extension_submitters" ON rate.extension_submitters; +CREATE POLICY "Rate worker read extension_submitters" ON rate.extension_submitters + FOR SELECT TO rate_worker USING (true); + +DROP POLICY IF EXISTS "Service role access extension_observations" ON rate.extension_observations; +CREATE POLICY "Service role access extension_observations" ON rate.extension_observations + FOR ALL TO service_role USING (true) WITH CHECK (true); +DROP POLICY IF EXISTS "Rate worker read extension_observations" ON rate.extension_observations; +CREATE POLICY "Rate worker read extension_observations" ON rate.extension_observations + FOR SELECT TO rate_worker USING (true); +-- The column grant above limits the update to status; the CHECK limits its values. +DROP POLICY IF EXISTS "Rate worker status extension_observations" ON rate.extension_observations; +CREATE POLICY "Rate worker status extension_observations" ON rate.extension_observations + FOR UPDATE TO rate_worker USING (true) WITH CHECK (true); + +-- 5. register_extension_submitter -------------------------------------------------------------- +-- The ingest key (42501, DETAIL ingest_key: its sha256 must equal the stored row, and no row +-- refuses every caller), a network hash (22023), then the limits (54000): at most 20 new +-- submitters per network hash in the last hour (DETAIL network_hash), then a global brake of +-- 600 in the last minute (DETAIL submitter). Each count runs under its advisory lock, so it is +-- exact: per network first, then the global one, always in that order. + +CREATE OR REPLACE FUNCTION rate.register_extension_submitter( + p_ingest_key text, + p_network_hash text +) +RETURNS uuid +LANGUAGE plpgsql +VOLATILE +SECURITY DEFINER +SET search_path = '' +AS $$ +DECLARE + v_id uuid; +BEGIN + -- The web's key has at least 32 characters; a shorter or missing one never reaches the hash. + IF p_ingest_key IS NULL OR char_length(p_ingest_key) < 32 THEN + RAISE EXCEPTION 'Extension ingest key refused' USING ERRCODE = '42501', + DETAIL = 'ingest_key'; + END IF; + IF (SELECT key_sha256 FROM rate.extension_ingest_secret WHERE id) IS DISTINCT FROM + encode(sha256(convert_to(coalesce(p_ingest_key, ''), 'UTF8')), 'hex') THEN + RAISE EXCEPTION 'Extension ingest key refused' USING ERRCODE = '42501', + DETAIL = 'ingest_key'; + END IF; + + IF p_network_hash IS NULL OR p_network_hash !~ '^[0-9a-f]{64}$' THEN + RAISE EXCEPTION 'Invalid network hash' USING ERRCODE = '22023', DETAIL = 'network_hash'; + END IF; + + PERFORM pg_catalog.pg_advisory_xact_lock( + pg_catalog.hashtextextended('rate.extension_submitters:' || p_network_hash, 0)); + IF (SELECT count(*) FROM rate.extension_submitters + WHERE network_hash = p_network_hash + AND created_at > now() - interval '1 hour') >= 20 THEN + RAISE EXCEPTION 'Too many extension submitters from this network, try again later' + USING ERRCODE = '54000', DETAIL = 'network_hash'; + END IF; + + PERFORM pg_catalog.pg_advisory_xact_lock( + pg_catalog.hashtextextended('rate.extension_submitters:register', 0)); + IF (SELECT count(*) FROM rate.extension_submitters + WHERE created_at > now() - interval '1 minute') >= 600 THEN + RAISE EXCEPTION 'Too many extension submitters registered, try again later' + USING ERRCODE = '54000', DETAIL = 'submitter'; + END IF; + + INSERT INTO rate.extension_submitters (network_hash) VALUES (p_network_hash) + RETURNING id INTO v_id; + RETURN v_id; +END; +$$; + +-- 6. submit_extension_observation -------------------------------------------------------------- +-- The ingest key (42501, DETAIL ingest_key), a network hash (22023, DETAIL network_hash), then +-- every rule of ExtensionObservation v1 (22023, DETAIL the field) in the order of +-- parseExtensionObservationV1 (packages/shared-data/src/rules/extension-observation.ts): +-- 'observation' for a payload that is not an object, is over 16 KB or has another key (its name +-- is never echoed), then each key in turn, its presence checked with its rule (a nullable key +-- must be present as null), then extractor.source and extractor.extension_version. Then the +-- submitter (P0002, DETAIL submitter_id) and the limits (54000): at most 120 per submitter and +-- 600 per network hash in the last hour. + +CREATE OR REPLACE FUNCTION rate.submit_extension_observation( + p_ingest_key text, + p_submitter_id uuid, + p_observation jsonb, + p_network_hash text +) +RETURNS jsonb +LANGUAGE plpgsql +VOLATILE +SECURITY DEFINER +SET search_path = '' +AS $$ +DECLARE + c_keys constant text[] := ARRAY['schema_version', 'marketplace', 'listing_url', + 'external_item_id', 'external_shop_id', 'sku_id', 'variant_label', 'price_vnd', + 'list_price_vnd', 'stock_state', 'stock_qty', 'observed_at', 'extractor']; + c_id_pattern constant text := '^[A-Za-z0-9_.-]{1,64}$'; + o jsonb := p_observation; + v_marketplace text; + v_url text; + v_host text; + v_host_ok boolean; + v_item text; + v_shop text; + v_sku text; + v_variant text; + v_price bigint; + v_list_price bigint; + v_stock_state text; + v_stock_qty integer; + v_observed_text text; + v_observed timestamptz; + v_extractor jsonb; + v_num numeric; + v_now timestamptz := now(); + v_id bigint; +BEGIN + -- The web's key has at least 32 characters; a shorter or missing one never reaches the hash. + IF p_ingest_key IS NULL OR char_length(p_ingest_key) < 32 THEN + RAISE EXCEPTION 'Extension ingest key refused' USING ERRCODE = '42501', + DETAIL = 'ingest_key'; + END IF; + IF (SELECT key_sha256 FROM rate.extension_ingest_secret WHERE id) IS DISTINCT FROM + encode(sha256(convert_to(coalesce(p_ingest_key, ''), 'UTF8')), 'hex') THEN + RAISE EXCEPTION 'Extension ingest key refused' USING ERRCODE = '42501', + DETAIL = 'ingest_key'; + END IF; + + IF p_network_hash IS NULL OR p_network_hash !~ '^[0-9a-f]{64}$' THEN + RAISE EXCEPTION 'Invalid network hash' USING ERRCODE = '22023', DETAIL = 'network_hash'; + END IF; + + IF o IS NULL OR jsonb_typeof(o) <> 'object' OR octet_length(o::text) > 16384 THEN + RAISE EXCEPTION 'Observation must be a JSON object' USING ERRCODE = '22023', + DETAIL = 'observation'; + END IF; + + -- Any other key is refused as 'observation': a caller's key name is never echoed. + IF EXISTS (SELECT 1 FROM jsonb_object_keys(o) AS k WHERE k <> ALL (c_keys)) THEN + RAISE EXCEPTION 'Unknown observation key' USING ERRCODE = '22023', DETAIL = 'observation'; + END IF; + + -- schema_version: exactly 1. + -- A number is read only once its type is known, so a string never reaches the cast. + v_num := CASE WHEN jsonb_typeof(o -> 'schema_version') = 'number' + THEN (o ->> 'schema_version')::numeric END; + IF v_num IS DISTINCT FROM 1 THEN + RAISE EXCEPTION 'Unsupported schema_version' USING ERRCODE = '22023', + DETAIL = 'schema_version'; + END IF; + + -- marketplace. + v_marketplace := CASE WHEN jsonb_typeof(o -> 'marketplace') = 'string' + THEN o ->> 'marketplace' END; + IF v_marketplace IS NULL OR v_marketplace NOT IN ('shopee', 'lazada', 'tiki', 'tiktok', + 'cellphones', 'tgdd', 'fptshop', 'hoangha', 'didongviet', 'phongvu') THEN + RAISE EXCEPTION 'Unsupported marketplace' USING ERRCODE = '22023', DETAIL = 'marketplace'; + END IF; + + -- listing_url: https, at most 2048 characters, no query, fragment, path parameter (;), space + -- (JavaScript's \s set) or control character; a lowercase host without user or port, from the marketplace's host table. + v_url := CASE WHEN jsonb_typeof(o -> 'listing_url') = 'string' THEN o ->> 'listing_url' END; + v_host := substring(v_url FROM '^https://([^/]*)'); + -- The host table (a CASE inside an IF condition would end the condition at its first THEN). + v_host_ok := CASE v_marketplace + WHEN 'shopee' THEN v_host = 'shopee.vn' + WHEN 'lazada' THEN v_host ~ '^([a-z0-9-]+\.)*lazada\.vn$' + WHEN 'tiki' THEN v_host = 'tiki.vn' + WHEN 'tiktok' THEN v_host IN ('www.tiktok.com', 'shop.tiktok.com') + WHEN 'cellphones' THEN v_host = 'cellphones.com.vn' + WHEN 'tgdd' THEN v_host ~ '^([a-z0-9-]+\.)*thegioididong\.com$' + WHEN 'fptshop' THEN v_host = 'fptshop.com.vn' + WHEN 'hoangha' THEN v_host = 'hoanghamobile.com' + WHEN 'didongviet' THEN v_host = 'didongviet.vn' + WHEN 'phongvu' THEN v_host = 'phongvu.vn' + ELSE false END; + IF v_url IS NULL OR char_length(v_url) > 2048 + OR v_url ~ '[?#;[:space:]\x01-\x1f\x7f-\x9f\xa0\x1680\x2000-\x200a\x2028\x2029\x202f\x205f\x3000\xfeff]' + OR v_host IS NULL OR v_host !~ '^[a-z0-9.-]+$' + OR v_host_ok IS NOT TRUE THEN + RAISE EXCEPTION 'Invalid listing_url' USING ERRCODE = '22023', DETAIL = 'listing_url'; + END IF; + + -- The ids: 1 to 64 of [A-Za-z0-9_.-]; the shop and SKU may be null. + v_item := CASE WHEN jsonb_typeof(o -> 'external_item_id') = 'string' + THEN o ->> 'external_item_id' END; + IF v_item IS NULL OR v_item !~ c_id_pattern THEN + RAISE EXCEPTION 'Invalid external_item_id' USING ERRCODE = '22023', + DETAIL = 'external_item_id'; + END IF; + IF NOT o ? 'external_shop_id' + OR COALESCE(jsonb_typeof(o -> 'external_shop_id'), 'null') NOT IN ('null', 'string') + OR (o ->> 'external_shop_id') !~ c_id_pattern THEN + RAISE EXCEPTION 'Invalid external_shop_id' USING ERRCODE = '22023', + DETAIL = 'external_shop_id'; + END IF; + v_shop := o ->> 'external_shop_id'; + IF NOT o ? 'sku_id' + OR COALESCE(jsonb_typeof(o -> 'sku_id'), 'null') NOT IN ('null', 'string') + OR (o ->> 'sku_id') !~ c_id_pattern THEN + RAISE EXCEPTION 'Invalid sku_id' USING ERRCODE = '22023', DETAIL = 'sku_id'; + END IF; + v_sku := o ->> 'sku_id'; + + -- variant_label: null, or 1 to 120 characters without a control character. + IF NOT o ? 'variant_label' + OR COALESCE(jsonb_typeof(o -> 'variant_label'), 'null') NOT IN ('null', 'string') + OR char_length(o ->> 'variant_label') NOT BETWEEN 1 AND 120 + OR (o ->> 'variant_label') ~ '[\x01-\x1f\x7f-\x9f]' THEN + RAISE EXCEPTION 'Invalid variant_label' USING ERRCODE = '22023', DETAIL = 'variant_label'; + END IF; + v_variant := o ->> 'variant_label'; + + -- price_vnd: an integer from 1,000 to 2,000,000,000. + v_num := CASE WHEN jsonb_typeof(o -> 'price_vnd') = 'number' + THEN (o ->> 'price_vnd')::numeric END; + IF v_num IS NULL OR v_num NOT BETWEEN 1000 AND 2000000000 OR v_num <> trunc(v_num) THEN + RAISE EXCEPTION 'Invalid price_vnd' USING ERRCODE = '22023', DETAIL = 'price_vnd'; + END IF; + v_price := v_num; + + -- list_price_vnd: null, or an integer from price_vnd to 2,000,000,000. + v_num := CASE WHEN jsonb_typeof(o -> 'list_price_vnd') = 'number' + THEN (o ->> 'list_price_vnd')::numeric END; + IF NOT o ? 'list_price_vnd' + OR COALESCE(jsonb_typeof(o -> 'list_price_vnd'), 'null') NOT IN ('null', 'number') + OR v_num NOT BETWEEN v_price AND 2000000000 OR v_num <> trunc(v_num) THEN + RAISE EXCEPTION 'Invalid list_price_vnd' USING ERRCODE = '22023', DETAIL = 'list_price_vnd'; + END IF; + v_list_price := v_num; + + -- stock_state and stock_qty. + v_stock_state := CASE WHEN jsonb_typeof(o -> 'stock_state') = 'string' + THEN o ->> 'stock_state' END; + IF v_stock_state IS NULL + OR v_stock_state NOT IN ('in_stock', 'low_stock', 'out_of_stock', 'unknown') THEN + RAISE EXCEPTION 'Invalid stock_state' USING ERRCODE = '22023', DETAIL = 'stock_state'; + END IF; + v_num := CASE WHEN jsonb_typeof(o -> 'stock_qty') = 'number' + THEN (o ->> 'stock_qty')::numeric END; + IF NOT o ? 'stock_qty' + OR COALESCE(jsonb_typeof(o -> 'stock_qty'), 'null') NOT IN ('null', 'number') + OR v_num NOT BETWEEN 0 AND 1000000 OR v_num <> trunc(v_num) THEN + RAISE EXCEPTION 'Invalid stock_qty' USING ERRCODE = '22023', DETAIL = 'stock_qty'; + END IF; + v_stock_qty := v_num; + + -- observed_at: ISO 8601 with a zone (seconds 00 to 59, so no leap second; an offset up to + -- 15:59, as Postgres parses it), at most 10 minutes ahead and 24 hours behind now(). + v_observed_text := CASE WHEN jsonb_typeof(o -> 'observed_at') = 'string' + THEN o ->> 'observed_at' END; + IF v_observed_text IS NULL OR v_observed_text !~ + '^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}(:[0-5]\d(\.\d{1,9})?)?(Z|[+-](0\d|1[0-5])(:?[0-5]\d)?)$' + THEN + RAISE EXCEPTION 'Invalid observed_at' USING ERRCODE = '22023', DETAIL = 'observed_at'; + END IF; + BEGIN + v_observed := v_observed_text::timestamptz; + EXCEPTION WHEN others THEN + RAISE EXCEPTION 'Invalid observed_at' USING ERRCODE = '22023', DETAIL = 'observed_at'; + END; + IF v_observed > v_now + interval '10 minutes' OR v_observed < v_now - interval '24 hours' THEN + RAISE EXCEPTION 'observed_at is out of range' USING ERRCODE = '22023', + DETAIL = 'observed_at'; + END IF; + + -- extractor: exactly {source, extension_version}. + v_extractor := o -> 'extractor'; + IF jsonb_typeof(v_extractor) IS DISTINCT FROM 'object' + OR v_extractor - 'source' - 'extension_version' <> '{}'::jsonb THEN + RAISE EXCEPTION 'Invalid extractor' USING ERRCODE = '22023', DETAIL = 'extractor'; + END IF; + IF jsonb_typeof(v_extractor -> 'source') IS DISTINCT FROM 'string' + OR (v_extractor ->> 'source') NOT IN ('json_ld', 'dom', 'mixed') THEN + RAISE EXCEPTION 'Invalid extractor.source' USING ERRCODE = '22023', + DETAIL = 'extractor.source'; + END IF; + IF jsonb_typeof(v_extractor -> 'extension_version') IS DISTINCT FROM 'string' + OR (v_extractor ->> 'extension_version') !~ '^[0-9A-Za-z.+-]{1,20}$' THEN + RAISE EXCEPTION 'Invalid extractor.extension_version' USING ERRCODE = '22023', + DETAIL = 'extractor.extension_version'; + END IF; + + -- The submitter, locked: one submitter's calls (and its deletion) run one at a time, so its + -- count below is exact. + PERFORM 1 FROM rate.extension_submitters WHERE id = p_submitter_id FOR UPDATE; + IF NOT FOUND THEN + RAISE EXCEPTION 'Unknown extension submitter' USING ERRCODE = 'P0002', + DETAIL = 'submitter_id'; + END IF; + + IF (SELECT count(*) FROM rate.extension_observations + WHERE submitter_id = p_submitter_id + AND received_at > v_now - interval '1 hour') >= 120 THEN + RAISE EXCEPTION 'Too many observations from this submitter, try again later' + USING ERRCODE = '54000', DETAIL = 'submitter_id'; + END IF; + + -- One network's calls run one at a time too (a transaction-scoped advisory lock). + PERFORM pg_catalog.pg_advisory_xact_lock( + pg_catalog.hashtextextended('rate.extension_observations:' || p_network_hash, 0)); + IF (SELECT count(*) FROM rate.extension_observations + WHERE network_hash = p_network_hash + AND received_at > v_now - interval '1 hour') >= 600 THEN + RAISE EXCEPTION 'Too many observations from this network, try again later' + USING ERRCODE = '54000', DETAIL = 'network_hash'; + END IF; + + INSERT INTO rate.extension_observations ( + submitter_id, schema_version, marketplace, listing_url, external_item_id, external_shop_id, + sku_id, variant_label, price_vnd, list_price_vnd, stock_state, stock_qty, observed_at, + extractor, received_at, network_hash, status) + VALUES ( + p_submitter_id, 1, v_marketplace, v_url, v_item, v_shop, + v_sku, v_variant, v_price, v_list_price, v_stock_state, v_stock_qty, v_observed, + v_extractor, v_now, p_network_hash, 'quarantined') + RETURNING id INTO v_id; + + RETURN jsonb_build_object('id', v_id, 'status', 'quarantined', 'received_at', v_now); +END; +$$; + +-- 7. delete_extension_submitter ---------------------------------------------------------------- +-- Deletes the submitter; its observations go by the cascade. Returns how many observations went; +-- an unknown or null id returns 0, so a call does not reveal whether an id exists. + +CREATE OR REPLACE FUNCTION rate.delete_extension_submitter(p_submitter_id uuid) +RETURNS integer +LANGUAGE plpgsql +VOLATILE +SECURITY DEFINER +SET search_path = '' +AS $$ +DECLARE + v_count integer; +BEGIN + -- Locked first: no observation of this submitter can be added between the count and the + -- delete. + PERFORM 1 FROM rate.extension_submitters WHERE id = p_submitter_id FOR UPDATE; + IF NOT FOUND THEN + RETURN 0; + END IF; + + SELECT count(*) INTO v_count FROM rate.extension_observations + WHERE submitter_id = p_submitter_id; + + DELETE FROM rate.extension_submitters WHERE id = p_submitter_id; + RETURN v_count; +END; +$$; + +-- 8. purge_extension_observations -------------------------------------------------------------- +-- Retention, ages from received_at and created_at. Meant to run daily as service_role. +-- network_hash cleared after p_network_hash_days (28), on observations and on submitters; +-- observations deleted after p_keep_days (180); +-- submitters without observations deleted after p_empty_submitter_days (7). +-- The empty submitters are locked first (SKIP LOCKED: one a submit holds waits for the next +-- run), then deleted under a fresh NOT EXISTS, so an observation committed in between keeps its +-- submitter instead of going with it by the cascade. + +CREATE OR REPLACE FUNCTION rate.purge_extension_observations( + p_network_hash_days integer DEFAULT 28, + p_keep_days integer DEFAULT 180, + p_empty_submitter_days integer DEFAULT 7 +) +RETURNS jsonb +LANGUAGE plpgsql +VOLATILE +SECURITY DEFINER +SET search_path = '' +AS $$ +DECLARE + v_hashes integer; + v_submitter_hashes integer; + v_observations integer; + v_submitters integer; + v_candidates uuid[]; +BEGIN + IF p_network_hash_days IS NULL OR p_network_hash_days < 1 THEN + RAISE EXCEPTION 'p_network_hash_days must be at least 1' USING ERRCODE = '22023', + DETAIL = 'p_network_hash_days'; + END IF; + IF p_keep_days IS NULL OR p_keep_days < 1 THEN + RAISE EXCEPTION 'p_keep_days must be at least 1' USING ERRCODE = '22023', + DETAIL = 'p_keep_days'; + END IF; + IF p_empty_submitter_days IS NULL OR p_empty_submitter_days < 1 THEN + RAISE EXCEPTION 'p_empty_submitter_days must be at least 1' USING ERRCODE = '22023', + DETAIL = 'p_empty_submitter_days'; + END IF; + + -- Lock order: submitters before observations, as delete_extension_submitter does (the + -- submitter row, then its observations through the cascade). The other order deadlocks with a + -- delete that runs in the middle of a purge. + UPDATE rate.extension_submitters SET network_hash = NULL + WHERE network_hash IS NOT NULL + AND created_at < now() - make_interval(days => p_network_hash_days); + GET DIAGNOSTICS v_submitter_hashes = ROW_COUNT; + + UPDATE rate.extension_observations SET network_hash = NULL + WHERE network_hash IS NOT NULL + AND received_at < now() - make_interval(days => p_network_hash_days); + GET DIAGNOSTICS v_hashes = ROW_COUNT; + + DELETE FROM rate.extension_observations + WHERE received_at < now() - make_interval(days => p_keep_days); + GET DIAGNOSTICS v_observations = ROW_COUNT; + + SELECT array_agg(c.id) INTO v_candidates FROM ( + SELECT s.id FROM rate.extension_submitters s + WHERE s.created_at < now() - make_interval(days => p_empty_submitter_days) + AND NOT EXISTS (SELECT 1 FROM rate.extension_observations o WHERE o.submitter_id = s.id) + FOR UPDATE OF s SKIP LOCKED + ) c; + + DELETE FROM rate.extension_submitters s + WHERE s.id = ANY (COALESCE(v_candidates, '{}')) + AND NOT EXISTS (SELECT 1 FROM rate.extension_observations o WHERE o.submitter_id = s.id); + GET DIAGNOSTICS v_submitters = ROW_COUNT; + + RETURN jsonb_build_object('network_hashes_cleared', v_hashes, + 'submitter_network_hashes_cleared', v_submitter_hashes, + 'observations_deleted', v_observations, + 'submitters_deleted', v_submitters); +END; +$$; + +-- 9. Execute grants ---------------------------------------------------------------------------- + +REVOKE EXECUTE ON FUNCTION rate.register_extension_submitter(text, text) + FROM PUBLIC, anon, authenticated, service_role, rate_worker; +REVOKE EXECUTE ON FUNCTION rate.submit_extension_observation(text, uuid, jsonb, text) + FROM PUBLIC, anon, authenticated, service_role, rate_worker; +REVOKE EXECUTE ON FUNCTION rate.delete_extension_submitter(uuid) + FROM PUBLIC, anon, authenticated, service_role, rate_worker; +REVOKE EXECUTE ON FUNCTION rate.purge_extension_observations(integer, integer, integer) + FROM PUBLIC, anon, authenticated, service_role, rate_worker; + +-- register and submit stay open to anon: the web calls them with the anon key plus the ingest +-- key. delete takes no ingest key, so erasure always works. +GRANT EXECUTE ON FUNCTION rate.register_extension_submitter(text, text) + TO anon, authenticated, service_role; +GRANT EXECUTE ON FUNCTION rate.submit_extension_observation(text, uuid, jsonb, text) + TO anon, authenticated, service_role; +GRANT EXECUTE ON FUNCTION rate.delete_extension_submitter(uuid) + TO anon, authenticated, service_role; +GRANT EXECUTE ON FUNCTION rate.purge_extension_observations(integer, integer, integer) + TO service_role; + +COMMIT; diff --git a/scripts/test-extension-observations-proposal.mjs b/scripts/test-extension-observations-proposal.mjs new file mode 100644 index 00000000..94aa3fe8 --- /dev/null +++ b/scripts/test-extension-observations-proposal.mjs @@ -0,0 +1,1177 @@ +#!/usr/bin/env node +// Replays the DOS-Me extension observations PROPOSAL (DOS/Rate#1606, M2 of +// Docs/Proposals/2026-10-28-extension-data-source.md, contract Revision 2) on the migration +// replay stack: the proposal twice; the ingest key (no row, a wrong key, the right key); every +// verification query; each payload, valid or refused, against the shared validator +// (parseExtensionObservationV1 in @repo/shared-data) with the same verdict and field; the +// unknown submitter; the limits at their exact thresholds; the delete cascade; the purge and its +// race with a submit; the table grants; the rollback twice and the proposal again. Nothing +// touches a real database. +// +// Usage: pnpm --filter @repo/shared-data build && node scripts/test-extension-observations-proposal.mjs +// (needs Docker) + +import assert from 'node:assert/strict'; +import { createHash } from 'node:crypto'; +import { readFileSync } from 'node:fs'; +import path from 'node:path'; +import { spawn, spawnSync } from 'node:child_process'; +import { pathToFileURL } from 'node:url'; + +import { repoRoot, startRateDb } from './lib/rate-db.mjs'; + +const rules = await import( + pathToFileURL( + path.join(repoRoot, 'packages', 'shared-data', 'dist', 'rules', 'index.js') + ).href +); +const parse = rules.parseExtensionObservationV1; +assert.equal( + typeof parse, + 'function', + 'parseExtensionObservationV1 is missing from @repo/shared-data: build it first' +); + +const read = (name) => + readFileSync(path.join(repoRoot, 'Docs', 'Proposals', name), 'utf8'); +const proposal = read('2026-11-04-dos-me-extension-observations.sql'); +const rollback = read('2026-11-04-dos-me-extension-observations-rollback.sql'); +const verify = read('2026-11-04-dos-me-extension-observations-verify.sql'); + +const container = `rate-extension-observations-proposal-${process.pid}`; +const removeContainer = () => + spawnSync('docker', ['rm', '--force', container], { timeout: 30_000 }); +for (const signal of ['SIGINT', 'SIGTERM']) { + process.on(signal, () => { + removeContainer(); + process.exit(130); + }); +} + +const checks = [ + ...verify.matchAll( + /^-- (V\d+) [^\n]*(?:\n--[^\n]*)*expect 0\n((?:SELECT|WITH)[\s\S]*?;)/gm + ), +].map(([, id, query]) => ({ id, query })); +assert.deepEqual( + checks.map((c) => c.id), + ['V1', 'V2', 'V3', 'V4', 'V5', 'V6', 'V7', 'V8', 'V9', 'V10', 'V11', 'V12'], + 'the verification lists its expect-0 queries' +); +const check = (id) => checks.find((c) => c.id === id).query; + +const q = (value) => `'${String(value).replaceAll("'", "''")}'`; +const j = (value) => `${q(JSON.stringify(value))}::jsonb`; +const sha = (text) => createHash('sha256').update(text, 'utf8').digest('hex'); +const ch = (code) => String.fromCodePoint(code); +// A test value made up here, not a secret: the replay sets its hash as the ingest key row. +const KEY = `local-replay-ingest-key-not-a-secret-${process.pid}`; +const NET_A = sha('net-a'); +const UNKNOWN = '00000000-0000-4000-8000-000000000000'; +let nets = 0; +const freshNet = () => sha(`net-${(nets += 1)}`); + +const MIN = 60_000; +const HOUR = 60 * MIN; +const pad = (n) => String(n).padStart(2, '0'); +// ISO 8601 at `ms`, written in the offset `off` minutes: sep ':' gives +07:00, '' gives +0700, +// null gives +07; offset 0 is Z. +const iso = (ms, { off = 0, sec = true, frac = '', sep = ':' } = {}) => { + const d = new Date(ms + off * MIN); + let s = `${d.getUTCFullYear()}-${pad(d.getUTCMonth() + 1)}-${pad(d.getUTCDate())}T${pad(d.getUTCHours())}:${pad(d.getUTCMinutes())}`; + if (sec) s += `:${pad(d.getUTCSeconds())}${frac}`; + if (off === 0) return `${s}Z`; + const a = Math.abs(off); + return `${s}${off > 0 ? '+' : '-'}${pad(Math.floor(a / 60))}${sep === null ? '' : `${sep}${pad(a % 60)}`}`; +}; +const ago = (ms) => iso(Date.now() - ms, { frac: '.000' }); + +const good = (over = {}) => ({ + schema_version: 1, + marketplace: 'shopee', + listing_url: 'https://shopee.vn/product/123456/7890123', + external_item_id: '7890123', + external_shop_id: '123456', + sku_id: null, + variant_label: 'Den, 128GB', + price_vnd: 10990000, + list_price_vnd: 12990000, + stock_state: 'in_stock', + stock_qty: null, + observed_at: new Date().toISOString(), + extractor: { source: 'json_ld', extension_version: '0.2.0' }, + ...over, +}); +const without = (key) => { + const o = good(); + delete o[key]; + return o; +}; + +try { + const { sql } = startRateDb({ name: container }); + // One psql session per call; the role and its errors stay in it. VERBOSITY verbose prints the + // SQLSTATE in front of the message. + const as = (role, query) => + sql(`\\set VERBOSITY verbose\nSET ROLE ${role};\n${query};\nRESET ROLE;`); + const refused = (role, query, code, detail, what) => + assert.throws( + () => as(role, query), + (err) => { + assert.match(err.message, new RegExp(`ERROR:\\s+${code}:`), what); + if (detail !== undefined) + assert.match( + err.message, + new RegExp(`DETAIL:\\s+${detail.replaceAll('.', '\\.')}\\r?\\n`), + what + ); + return true; + }, + what + ); + const checkAll = (when) => { + for (const { id, query } of checks) + assert.equal(sql(query), '0', `${id} ${when}`); + }; + const lit = (v) => (v === null ? 'NULL' : q(v)); + const registerSql = (net, key = KEY) => + `SELECT rate.register_extension_submitter(${lit(key)}, ${lit(net)})`; + const register = (net = freshNet(), role = 'anon') => + as(role, registerSql(net)); + const submitSql = (id, obs, net = NET_A, key = KEY) => + `SELECT rate.submit_extension_observation(${lit(key)}, ${lit(id)}, ${ + obs === undefined ? 'NULL' : j(obs) + }, ${lit(net)})::text`; + const submit = (id, obs, net = NET_A, role = 'anon') => + JSON.parse(as(role, submitSql(id, obs, net))); + const count = (where = 'true', table = 'rate.extension_observations') => + Number(sql(`SELECT count(*) FROM ${table} WHERE ${where}`)); + // Rows written straight as postgres, received (and observed) `age` ago. + const backfill = (id, n, age, net = null) => + sql(`INSERT INTO rate.extension_observations (submitter_id, schema_version, marketplace, + listing_url, external_item_id, price_vnd, stock_state, observed_at, extractor, + received_at, network_hash) + SELECT ${q(id)}, 1, 'shopee', 'https://shopee.vn/p', 'i' || g, 100000, 'unknown', + now() - ${q(age)}::interval, '{"source": "dom", "extension_version": "1"}', + now() - ${q(age)}::interval, ${lit(net)} + FROM generate_series(1, ${n}) g`); + + // 1. The proposal twice. + sql(proposal); + sql(proposal); + + // 2. The ingest key: no row refuses every caller; then a wrong, empty or null key refuses and + // the right one passes. delete takes no key. + refused( + 'anon', + registerSql(NET_A), + '42501', + 'ingest_key', + 'register without a key row' + ); + refused( + 'anon', + submitSql(UNKNOWN, good()), + '42501', + 'ingest_key', + 'submit without a key row' + ); + assert.equal( + count('true', 'rate.extension_ingest_secret'), + 0, + 'the migration sets no key' + ); + sql( + `INSERT INTO rate.extension_ingest_secret (key_sha256) VALUES (${q(sha(KEY))})` + ); + refused( + 'postgres', + `INSERT INTO rate.extension_ingest_secret (key_sha256) VALUES (${q(sha('x'))})`, + '23505', + undefined, + 'one row only' + ); + refused( + 'postgres', + `INSERT INTO rate.extension_ingest_secret (id, key_sha256) + VALUES (false, ${q(sha('x'))})`, + '23514', + undefined, + 'id is always true' + ); + refused( + 'postgres', + `UPDATE rate.extension_ingest_secret SET key_sha256 = 'ABC'`, + '23514', + undefined, + 'the hash format is checked' + ); + for (const key of ['wrong', '', null, KEY.toUpperCase(), sha(KEY)]) { + refused( + 'anon', + registerSql(NET_A, key), + '42501', + 'ingest_key', + `register with ${key}` + ); + refused( + 'anon', + submitSql(UNKNOWN, good(), NET_A, key), + '42501', + 'ingest_key', + `submit with ${key}` + ); + } + // The empty key's hash cannot be stored; a key under 32 characters is refused even when its + // hash is the stored one. + refused( + 'postgres', + `UPDATE rate.extension_ingest_secret SET key_sha256 = ${q(sha(''))}`, + '23514', + undefined, + 'the empty key hash is refused' + ); + const shortKey = 'k'.repeat(31); + sql( + `UPDATE rate.extension_ingest_secret SET key_sha256 = ${q(sha(shortKey))}` + ); + refused( + 'anon', + registerSql(NET_A, shortKey), + '42501', + 'ingest_key', + 'a 31-character key' + ); + refused( + 'anon', + submitSql(UNKNOWN, good(), NET_A, shortKey), + '42501', + 'ingest_key', + 'a 31-character key on submit' + ); + sql(`UPDATE rate.extension_ingest_secret SET key_sha256 = ${q(sha(KEY))}`); + assert.ok(KEY.length >= 32, 'the replay key is long enough'); + + // 3. Every check, and a check fails when its rule breaks. + checkAll('after the apply'); + sql(verify); + sql('GRANT SELECT ON rate.extension_observations TO anon'); + assert.notEqual(sql(check('V2')), '0', 'V2 sees a grant to anon'); + sql('REVOKE SELECT ON rate.extension_observations FROM anon'); + sql('GRANT SELECT ON rate.extension_ingest_secret TO rate_worker'); + assert.notEqual(sql(check('V11')), '0', 'V11 sees a grant on the key'); + sql('REVOKE SELECT ON rate.extension_ingest_secret FROM rate_worker'); + sql('ALTER TABLE rate.extension_observations FORCE ROW LEVEL SECURITY'); + assert.notEqual(sql(check('V1')), '0', 'V1 sees forced row level security'); + sql('ALTER TABLE rate.extension_observations NO FORCE ROW LEVEL SECURITY'); + checkAll('after the probes'); + + // 4. Register: the right key and a network hash; anon and authenticated; not rate_worker. + const s1 = register(); + assert.match(s1, /^[0-9a-f-]{36}$/); + const sAuth = register(freshNet(), 'authenticated'); + assert.notEqual(s1, sAuth); + assert.equal( + sql( + `SELECT network_hash FROM rate.extension_submitters WHERE id = ${q(s1)}` + ), + sha(`net-${nets - 1}`), + 'the submitter keeps its network hash' + ); + refused( + 'rate_worker', + registerSql(NET_A), + '42501', + undefined, + 'rate_worker cannot register' + ); + for (const net of [ + null, + 'abc', + NET_A.toUpperCase(), + NET_A.slice(1), + `${NET_A}0`, + '', + ]) { + refused( + 'anon', + registerSql(net), + '22023', + 'network_hash', + `register network_hash ${net}` + ); + refused( + 'anon', + submitSql(s1, good(), net), + '22023', + 'network_hash', + `submit network_hash ${net}` + ); + } + + // 5. A valid observation: the row, the answer. + const r1 = submit(s1, good()); + assert.deepEqual(Object.keys(r1).sort(), ['id', 'received_at', 'status']); + assert.equal(r1.status, 'quarantined'); + assert.equal(typeof r1.id, 'number'); + assert.equal( + sql(`SELECT submitter_id || '|' || marketplace || '|' || price_vnd || '|' || list_price_vnd + || '|' || coalesce(sku_id, '-') || '|' || status || '|' || network_hash || '|' + || (received_at = ${q(r1.received_at)}::timestamptz) || '|' || extractor::text + FROM rate.extension_observations WHERE id = ${r1.id}`), + `${s1}|shopee|10990000|12990000|-|quarantined|${NET_A}|true|{"source": "json_ld", "extension_version": "0.2.0"}` + ); + assert.equal( + submit(sAuth, good(), NET_A, 'authenticated').status, + 'quarantined' + ); + assert.equal(submit(s1, good(), NET_A, 'service_role').status, 'quarantined'); + + // 6. Each payload against the shared validator: the same verdict and the same field. The RPC + // writes the valid ones and nothing for the others. + const now = Date.now(); + const emoji = ch(0x1f600); + const valid = [ + {}, + { + marketplace: 'lazada', + listing_url: 'https://www.lazada.vn/products/x-i1.html', + }, + { marketplace: 'lazada', listing_url: 'https://lazada.vn/products/x' }, + { marketplace: 'lazada', listing_url: 'https://a.b-c.lazada.vn/x' }, + { marketplace: 'tiki', listing_url: 'https://tiki.vn/x-p1.html' }, + { + marketplace: 'tiktok', + listing_url: 'https://www.tiktok.com/view/product/1', + }, + { + marketplace: 'tiktok', + listing_url: 'https://shop.tiktok.com/view/product/1', + }, + { + marketplace: 'cellphones', + listing_url: 'https://cellphones.com.vn/x.html', + }, + { + marketplace: 'tgdd', + listing_url: 'https://www.thegioididong.com/dtdd/x', + }, + { marketplace: 'tgdd', listing_url: 'https://thegioididong.com/dtdd/x' }, + { + marketplace: 'fptshop', + listing_url: 'https://fptshop.com.vn/dien-thoai/x', + }, + { marketplace: 'hoangha', listing_url: 'https://hoanghamobile.com/x' }, + { marketplace: 'didongviet', listing_url: 'https://didongviet.vn/x.html' }, + { marketplace: 'phongvu', listing_url: 'https://phongvu.vn/x' }, + { listing_url: 'https://shopee.vn' }, + { + listing_url: `https://shopee.vn/${'a'.repeat(2048 - 'https://shopee.vn/'.length)}`, + }, + { + listing_url: `https://shopee.vn/${emoji.repeat(2048 - 'https://shopee.vn/'.length)}`, + }, + { listing_url: `https://shopee.vn/${ch(0x200b)}x` }, + { external_shop_id: null, variant_label: null, list_price_vnd: null }, + { + price_vnd: 1000, + list_price_vnd: 1000, + stock_qty: 0, + stock_state: 'low_stock', + }, + { price_vnd: 2000000000, list_price_vnd: 2000000000, stock_qty: 1000000 }, + { price_vnd: 1000.0, schema_version: 1.0 }, + { + external_item_id: 'A'.repeat(64), + sku_id: 'sku_1.2-3', + variant_label: 'x'.repeat(120), + }, + { variant_label: emoji.repeat(120) }, + { variant_label: `Mau ${ch(0x111)}en ${ch(0x200b)}` }, + { + observed_at: iso(now - 23 * HOUR - 59 * MIN), + stock_state: 'out_of_stock', + }, + { observed_at: iso(now + 9 * MIN + 30_000), stock_state: 'unknown' }, + { observed_at: iso(now - HOUR, { off: 420 }) }, + { observed_at: iso(now - HOUR, { off: 420, sep: '' }) }, + { observed_at: iso(now - HOUR, { off: 420, sep: null }) }, + { observed_at: iso(now - HOUR, { off: -300 }) }, + { observed_at: iso(now - HOUR, { off: 345 }) }, + { observed_at: iso(now - HOUR, { off: 959 }) }, + { observed_at: iso(now - HOUR, { off: -900 }) }, + { observed_at: iso(now - HOUR, { sec: false }) }, + { observed_at: iso(now - HOUR, { frac: '.123456789' }) }, + { observed_at: iso(now - HOUR, { frac: '.5' }) }, + { extractor: { source: 'mixed', extension_version: '1.0.0-beta+2' } }, + { extractor: { source: 'dom', extension_version: 'v'.repeat(20) } }, + ].map((over) => ['ok', good(over)]); + const hourAgo = iso(now - HOUR); + const bad = [ + ['observation', []], + ['observation', 'x'], + ['observation', null], + ['observation', 5], + ['observation', { ...good(), rating: 5 }], + ['observation', { ...good(), cookies: 'a=b' }], + ['observation', { ...good(), ["x'; DROP TABLE x; --"]: 1 }], + ['observation', { ...without('schema_version'), referrer: 'https://x' }], + ['schema_version', good({ schema_version: 2 })], + ['schema_version', good({ schema_version: '1' })], + ['schema_version', good({ schema_version: null })], + ['marketplace', good({ marketplace: 'amazon' })], + ['marketplace', good({ marketplace: 'Shopee' })], + ['marketplace', good({ marketplace: 'thegioididong' })], + ['marketplace', good({ marketplace: 'hoanghamobile' })], + ['marketplace', good({ marketplace: null })], + ['listing_url', good({ listing_url: 'http://shopee.vn/x' })], + ['listing_url', good({ listing_url: 'https://shopee.vn/x?sp_atk=1' })], + ['listing_url', good({ listing_url: 'https://shopee.vn/x#reviews' })], + ['listing_url', good({ listing_url: 'https://shopee.vn/x;jsessionid=1' })], + ['listing_url', good({ listing_url: 'https://shopee.vn;x' })], + ['listing_url', good({ listing_url: 'https://shopee.vn/x y' })], + ['listing_url', good({ listing_url: 'https://shopee.vn/x\ny' })], + ['listing_url', good({ listing_url: 'https://shopee.vn/x\ty' })], + ...[ + 0xa0, 0x85, 0x1680, 0x2000, 0x200a, 0x2028, 0x2029, 0x202f, 0x205f, + 0x3000, 0xfeff, + ].map((c) => [ + 'listing_url', + good({ listing_url: `https://shopee.vn/x${ch(c)}y` }), + ]), + ['listing_url', good({ listing_url: 'https://Shopee.vn/x' })], + ['listing_url', good({ listing_url: 'https://shopee.vn:443/x' })], + ['listing_url', good({ listing_url: 'https://user@shopee.vn/x' })], + ['listing_url', good({ listing_url: 'https://shopee.vn./x' })], + ['listing_url', good({ listing_url: 'https:///shopee.vn/x' })], + ['listing_url', good({ listing_url: 'https://shopee.vn.evil.com/x' })], + ['listing_url', good({ listing_url: 'https://evil.com/shopee.vn' })], + ['listing_url', good({ listing_url: 'https://www.shopee.vn/x' })], + ['listing_url', good({ listing_url: 'https://lazada.vn/x' })], + [ + 'listing_url', + good({ marketplace: 'lazada', listing_url: 'https://xlazada.vn/x' }), + ], + [ + 'listing_url', + good({ marketplace: 'tiktok', listing_url: 'https://tiktok.com/x' }), + ], + [ + 'listing_url', + good({ marketplace: 'tiktok', listing_url: 'https://m.tiktok.com/x' }), + ], + [ + 'listing_url', + good({ listing_url: `https://shopee.vn/${'a'.repeat(2031)}` }), + ], + [ + 'listing_url', + good({ listing_url: `https://shopee.vn/${emoji.repeat(2031)}` }), + ], + ['listing_url', good({ listing_url: 'https://shopee.vn\\@evil.com/x' })], + ['listing_url', good({ listing_url: 5 })], + ['listing_url', good({ listing_url: null })], + ['external_item_id', good({ external_item_id: '' })], + ['external_item_id', good({ external_item_id: 'a'.repeat(65) })], + ['external_item_id', good({ external_item_id: 'a b' })], + ['external_item_id', good({ external_item_id: 'a/b' })], + ['external_item_id', good({ external_item_id: 'a\n' })], + ['external_item_id', good({ external_item_id: 123 })], + ['external_shop_id', good({ external_shop_id: '' })], + ['external_shop_id', good({ external_shop_id: 'x y' })], + ['external_shop_id', good({ external_shop_id: 5 })], + ['sku_id', good({ sku_id: 'a'.repeat(65) })], + ['sku_id', good({ sku_id: ch(0xe1) })], + ['variant_label', good({ variant_label: '' })], + ['variant_label', good({ variant_label: 'x'.repeat(121) })], + ['variant_label', good({ variant_label: emoji.repeat(121) })], + ['variant_label', good({ variant_label: 'a\nb' })], + ['variant_label', good({ variant_label: `a${ch(0x85)}b` })], + ['variant_label', good({ variant_label: `a${ch(0x7f)}b` })], + ['variant_label', good({ variant_label: 5 })], + ['price_vnd', good({ price_vnd: 999 })], + ['price_vnd', good({ price_vnd: 2000000001 })], + ['price_vnd', good({ price_vnd: 1000.5 })], + ['price_vnd', good({ price_vnd: '10990000' })], + ['price_vnd', good({ price_vnd: null })], + ['list_price_vnd', good({ list_price_vnd: 10989999 })], + ['list_price_vnd', good({ list_price_vnd: 2000000001 })], + ['list_price_vnd', good({ list_price_vnd: 11000000.5 })], + ['list_price_vnd', good({ list_price_vnd: '12990000' })], + ['stock_state', good({ stock_state: 'sold_out' })], + ['stock_state', good({ stock_state: null })], + ['stock_qty', good({ stock_qty: -1 })], + ['stock_qty', good({ stock_qty: 1000001 })], + ['stock_qty', good({ stock_qty: 1.5 })], + ['stock_qty', good({ stock_qty: '3' })], + ['observed_at', good({ observed_at: iso(now - 24 * HOUR - 60_000) })], + ['observed_at', good({ observed_at: iso(now + 10 * MIN + 60_000) })], + ['observed_at', good({ observed_at: 'yesterday' })], + ['observed_at', good({ observed_at: 'now' })], + ['observed_at', good({ observed_at: '2026-13-01T00:00:00Z' })], + ['observed_at', good({ observed_at: hourAgo.slice(0, 19) })], + ['observed_at', good({ observed_at: hourAgo.replace('T', ' ') })], + ['observed_at', good({ observed_at: hourAgo.replace('Z', 'z') })], + ['observed_at', good({ observed_at: hourAgo.replace(/:\d\dZ$/, ':61Z') })], + ['observed_at', good({ observed_at: hourAgo.replace(/T\d\d/, 'T25') })], + ['observed_at', good({ observed_at: hourAgo.replace('Z', '+24:00') })], + ['observed_at', good({ observed_at: hourAgo.replace(/:\d\dZ$/, ':60Z') })], + ['observed_at', good({ observed_at: hourAgo.replace('Z', '+16:00') })], + ['observed_at', good({ observed_at: hourAgo.replace('Z', '-16:00') })], + ['observed_at', good({ observed_at: hourAgo.replace('Z', '+1560') })], + ['observed_at', good({ observed_at: hourAgo.replace('Z', '+05:60') })], + [ + 'observed_at', + good({ observed_at: `${hourAgo.slice(0, 19)}.1234567890Z` }), + ], + ['observed_at', good({ observed_at: now })], + ['observed_at', good({ observed_at: null })], + ['extractor', good({ extractor: 'dom' })], + ['extractor', good({ extractor: null })], + ['extractor', good({ extractor: [] })], + [ + 'extractor', + good({ extractor: { source: 'dom', extension_version: '1', x: 1 } }), + ], + [ + 'extractor.source', + good({ extractor: { source: 'ai', extension_version: '1' } }), + ], + ['extractor.source', good({ extractor: { extension_version: '1' } })], + ['extractor.extension_version', good({ extractor: { source: 'dom' } })], + [ + 'extractor.extension_version', + good({ extractor: { source: 'dom', extension_version: '' } }), + ], + [ + 'extractor.extension_version', + good({ extractor: { source: 'dom', extension_version: 'v'.repeat(21) } }), + ], + [ + 'extractor.extension_version', + good({ extractor: { source: 'dom', extension_version: 'v 1' } }), + ], + [ + 'extractor.extension_version', + good({ extractor: { source: 'dom', extension_version: 1 } }), + ], + // Every key must be present, a nullable one as null; each is named by its own check. + ...Object.keys(good()).map((key) => [key, without(key)]), + ]; + // The RPC's verdict: 'ok', or the field of its 22023. + const sParity = register(); + const rpcVerdict = (obs) => { + try { + submit(sParity, obs); + return 'ok'; + } catch (err) { + const m = /ERROR:\s+(\w+):[\s\S]*?DETAIL:\s+(\S+)\r?\n/.exec(err.message); + assert.ok(m, err.message); + assert.equal(m[1], '22023', err.message); + return m[2]; + } + }; + const before = count(); + const disagree = []; + for (const [expected, obs] of [...valid, ...bad]) { + const ts = parse(obs, Date.now()); + const tsVerdict = ts.ok ? 'ok' : ts.field; + const db = rpcVerdict(obs); + const what = JSON.stringify(obs).slice(0, 160); + if (tsVerdict !== db) + disagree.push(`${what}: shared-data ${tsVerdict}, RPC ${db}`); + assert.equal(db, expected, `RPC on ${what}`); + } + assert.deepEqual( + disagree, + [], + 'the RPC and parseExtensionObservationV1 agree' + ); + assert.equal( + count() - before, + valid.length, + 'only the valid payloads are written' + ); + // Where they differ on purpose: the 16 KB cap is the RPC's alone (the web caps the body). + const big = good({ variant_label: 'x'.repeat(17000) }); + assert.equal(parse(big, Date.now()).field, 'variant_label'); + assert.equal(rpcVerdict(big), 'observation'); + // A lone surrogate: shared-data names the field; jsonb refuses the text itself. + const lone = good({ + listing_url: `https://shopee.vn/x${String.fromCharCode(0xd800)}`, + }); + assert.equal(parse(lone, Date.now()).field, 'listing_url'); + refused( + 'anon', + submitSql(sParity, lone), + '22P02', + undefined, + 'jsonb refuses a lone surrogate' + ); + // A SQL NULL payload, an unknown submitter and a null one. + refused( + 'anon', + submitSql(s1, undefined), + '22023', + 'observation', + 'a NULL payload' + ); + refused( + 'anon', + submitSql(UNKNOWN, good()), + 'P0002', + 'submitter_id', + 'unknown submitter' + ); + refused( + 'anon', + submitSql(null, good()), + 'P0002', + 'submitter_id', + 'null submitter' + ); + assert.equal(count() - before, valid.length, 'refused calls write nothing'); + + // 7. The limits at their exact thresholds (54000). + // 120 an hour per submitter: 119 in the hour and 50 older, then the 120th passes. + const sLimit = register(); + backfill(sLimit, 119, '30 minutes'); + backfill(sLimit, 50, '61 minutes'); + submit(sLimit, good()); + refused( + 'anon', + submitSql(sLimit, good()), + '54000', + 'submitter_id', + '121st in an hour' + ); + assert.equal(count(`submitter_id = ${q(sLimit)}`), 170); + // 600 an hour per network: 599 in the hour and 100 older, then the 600th passes. + const NET_C = freshNet(); + const sBulk = register(); + backfill(sBulk, 599, '30 minutes', NET_C); + backfill(sBulk, 100, '61 minutes', NET_C); + const sNet = register(); + assert.equal( + submit(sNet, good(), NET_C).status, + 'quarantined', + '600th from a network' + ); + refused( + 'anon', + submitSql(sNet, good(), NET_C), + '54000', + 'network_hash', + '601st from a network' + ); + assert.equal( + submit(sNet, good(), NET_A).status, + 'quarantined', + 'another network passes' + ); + // 20 registers an hour per network: 19 in the hour and 5 older, then the 20th passes. + const NET_D = freshNet(); + sql(`INSERT INTO rate.extension_submitters (created_at, network_hash) + SELECT now() - interval '30 minutes', ${q(NET_D)} FROM generate_series(1, 19); + INSERT INTO rate.extension_submitters (created_at, network_hash) + SELECT now() - interval '61 minutes', ${q(NET_D)} FROM generate_series(1, 5)`); + assert.match( + register(NET_D), + /^[0-9a-f-]{36}$/, + '20th register from a network' + ); + refused( + 'anon', + registerSql(NET_D), + '54000', + 'network_hash', + '21st register from a network' + ); + // 600 registers a minute in all: fill to 599, then the 600th passes and the 601st is refused. + // The test's own registers age out of the minute as it runs, so they are moved out first. + sql(`UPDATE rate.extension_submitters SET created_at = now() - interval '2 minutes' + WHERE created_at > now() - interval '2 minutes'; + INSERT INTO rate.extension_submitters (created_at) + SELECT now() - interval '20 seconds' FROM generate_series(1, 599)`); + assert.match(register(), /^[0-9a-f-]{36}$/, '600th register in a minute'); + refused( + 'anon', + registerSql(freshNet()), + '54000', + 'submitter', + '601st register in a minute' + ); + sql(`UPDATE rate.extension_submitters SET created_at = now() - interval '2 minutes' + WHERE created_at > now() - interval '1 minute'`); + assert.match(register(), /^[0-9a-f-]{36}$/, 'the brake lifts after a minute'); + + // 8. Delete: no key; the submitter and its observations; the count; 0 for an unknown or null id. + const sDel = register(); + for (let i = 0; i < 3; i += 1) submit(sDel, good()); + assert.equal( + as('anon', `SELECT rate.delete_extension_submitter(${q(sDel)})`), + '3' + ); + assert.equal(count(`submitter_id = ${q(sDel)}`), 0); + assert.equal(count(`id = ${q(sDel)}`, 'rate.extension_submitters'), 0); + assert.equal( + as('anon', `SELECT rate.delete_extension_submitter(${q(sDel)})`), + '0' + ); + assert.equal( + as('anon', `SELECT rate.delete_extension_submitter(${q(UNKNOWN)})`), + '0' + ); + assert.equal( + as('authenticated', 'SELECT rate.delete_extension_submitter(NULL)'), + '0' + ); + const sEmpty = register(); + assert.equal( + as('authenticated', `SELECT rate.delete_extension_submitter(${q(sEmpty)})`), + '0' + ); + assert.equal(count(`id = ${q(sEmpty)}`, 'rate.extension_submitters'), 0); + refused( + 'anon', + submitSql(sDel, good()), + 'P0002', + 'submitter_id', + 'a deleted submitter' + ); + const sCascade = register(); + submit(sCascade, good()); + sql(`DELETE FROM rate.extension_submitters WHERE id = ${q(sCascade)}`); + assert.equal( + count(`submitter_id = ${q(sCascade)}`), + 0, + 'the cascade holds for service_role' + ); + + // 9. Purge: hashes after 28 days (observations and submitters), rows after 180, submitters + // without observations after 7. + const [p1, p2, p3, p4, p5, p6] = Array.from({ length: 6 }, () => register()); + const age = (ids, days) => + sql(`UPDATE rate.extension_submitters SET created_at = now() - interval '${days} days' + WHERE id IN (${ids.map(q).join(', ')})`); + age([p1, p2, p3], 200); + age([p4], 10); + age([p5], 3); + age([p6], 30); + backfill(p1, 1, '200 days', NET_A); + backfill(p1, 1, '30 days', NET_A); + backfill(p1, 1, '10 days', NET_A); + backfill(p3, 1, '190 days'); + backfill(p6, 1, '1 day', NET_A); + const total = count(); + for (const role of ['anon', 'authenticated', 'rate_worker']) + refused( + role, + 'SELECT rate.purge_extension_observations()', + '42501', + undefined, + `${role} purge` + ); + refused( + 'service_role', + 'SELECT rate.purge_extension_observations(0)', + '22023', + 'p_network_hash_days', + 'purge hash days' + ); + refused( + 'service_role', + 'SELECT rate.purge_extension_observations(28, NULL)', + '22023', + 'p_keep_days', + 'purge keep days' + ); + refused( + 'service_role', + 'SELECT rate.purge_extension_observations(28, 180, 0)', + '22023', + 'p_empty_submitter_days', + 'purge empty days' + ); + assert.deepEqual( + JSON.parse( + as('service_role', 'SELECT rate.purge_extension_observations()::text') + ), + { + network_hashes_cleared: 2, + submitter_network_hashes_cleared: 4, + observations_deleted: 2, + submitters_deleted: 3, + } + ); + assert.equal(count(), total - 2); + assert.equal( + sql(`SELECT string_agg(to_char(now() - received_at, 'DD') || ':' || (network_hash IS NOT NULL), + ',' ORDER BY received_at) FROM rate.extension_observations WHERE submitter_id = ${q(p1)}`), + '30:false,10:true' + ); + assert.equal( + sql(`SELECT string_agg(left(id::text, 8) || ':' || (network_hash IS NOT NULL), ',' + ORDER BY created_at) FROM rate.extension_submitters + WHERE id IN (${[p1, p2, p3, p4, p5, p6].map(q).join(', ')})`), + `${p1.slice(0, 8)}:false,${p6.slice(0, 8)}:false,${p5.slice(0, 8)}:true`, + 'p2 (never sent), p3 (emptied) and p4 (empty for 10 days) go; p1 and p6 keep rows, p5 is young' + ); + assert.deepEqual( + JSON.parse( + as('service_role', 'SELECT rate.purge_extension_observations()::text') + ), + { + network_hashes_cleared: 0, + submitter_network_hashes_cleared: 0, + observations_deleted: 0, + submitters_deleted: 0, + }, + 'a second purge finds nothing' + ); + // The race: a submit for an empty, purgeable submitter is still open when the purge runs. The + // purge skips the submitter it holds; the observation and its submitter stay. + const p7 = register(); + age([p7], 10); + const racing = spawn( + 'docker', + [ + 'exec', + '-i', + container, + 'psql', + '-XqAt', + '-v', + 'ON_ERROR_STOP=1', + '-U', + 'postgres', + ], + { stdio: ['pipe', 'pipe', 'pipe'] } + ); + let racingOut = ''; + racing.stdout.on('data', (d) => (racingOut += d)); + racing.stderr.on('data', (d) => (racingOut += d)); + const racingDone = new Promise((resolve) => racing.on('close', resolve)); + racing.stdin.end(`BEGIN; SET LOCAL ROLE anon; + SELECT 'submitted:' || (${submitSql(p7, good()).slice(7)} IS NOT NULL); + SELECT pg_sleep(3); COMMIT;`); + const deadline = Date.now() + 10_000; + while ( + count(`submitter_id = ${q(p7)}`) === 0 && + sql(`SELECT count(*) FROM pg_locks l JOIN pg_class c ON c.oid = l.relation + WHERE c.relname = 'extension_submitters' AND l.mode = 'RowShareLock' + AND l.pid <> pg_backend_pid()`) === '0' + ) { + assert.ok(Date.now() < deadline, 'the racing submit started'); + await new Promise((resolve) => setTimeout(resolve, 200)); + } + const raced = JSON.parse( + as('service_role', 'SELECT rate.purge_extension_observations()::text') + ); + assert.equal(await racingDone, 0, racingOut); + assert.match(racingOut, /submitted:true/); + assert.equal( + raced.submitters_deleted, + 0, + 'the purge skipped the held submitter' + ); + assert.equal( + count(`id = ${q(p7)}`, 'rate.extension_submitters'), + 1, + 'p7 stays' + ); + assert.equal(count(`submitter_id = ${q(p7)}`), 1, 'its observation stays'); + + // The lock order: a delete in the middle of a purge must not deadlock. A third session holds + // an old observation X and an old submitter Y, so the purge stops partway; then the user + // deletes S, whose old observation O the purge also clears. Purge and delete both take the + // submitter before its observations, so the delete waits for the purge and both finish. + const background = (input) => { + const child = spawn( + 'docker', + [ + 'exec', + '-i', + container, + 'psql', + '-XqAt', + '-v', + 'ON_ERROR_STOP=1', + '-U', + 'postgres', + ], + { stdio: ['pipe', 'pipe', 'pipe'] } + ); + let out = ''; + child.stdout.on('data', (d) => (out += d)); + child.stderr.on('data', (d) => (out += d)); + child.stdin.end(input); + return new Promise((resolve) => + child.on('close', (code) => resolve({ code, out })) + ); + }; + // Async between polls, so the background sessions' pipes keep flowing. + const waitFor = async (query, what) => { + const until = Date.now() + 15_000; + while (sql(query) === '0') { + assert.ok(Date.now() < until, what); + await new Promise((resolve) => setTimeout(resolve, 200)); + } + }; + const [dS, dY, dZ] = [register(), register(), register()]; + age([dS, dY, dZ], 30); + backfill(dS, 1, '30 days', NET_A); // O + backfill(dZ, 1, '30 days', NET_A); // X + const holder = background(`BEGIN; + SELECT 1 FROM rate.extension_observations WHERE submitter_id = ${q(dZ)} FOR UPDATE; + SELECT 1 FROM rate.extension_submitters WHERE id = ${q(dY)} FOR UPDATE; + SELECT pg_sleep(4); COMMIT;`); + await waitFor( + `SELECT count(*) FROM pg_stat_activity WHERE query LIKE '%pg_sleep(4)%' + AND pid <> pg_backend_pid() AND wait_event = 'PgSleep'`, + 'the holder took its locks' + ); + const purging = background( + `SET ROLE service_role; SELECT rate.purge_extension_observations()::text;` + ); + await waitFor( + `SELECT count(*) FROM pg_stat_activity WHERE query LIKE '%purge_extension_observations%' + AND pid <> pg_backend_pid() AND wait_event_type = 'Lock'`, + 'the purge waits on the holder' + ); + const deleting = background( + `SET ROLE anon; SELECT rate.delete_extension_submitter(${q(dS)});` + ); + const [held, purged, deleted] = await Promise.all([ + holder, + purging, + deleting, + ]); + assert.equal(held.code, 0, held.out); + assert.equal(purged.code, 0, `the purge finished: ${purged.out}`); + assert.equal(deleted.code, 0, `the delete finished: ${deleted.out}`); + assert.doesNotMatch(`${purged.out}${deleted.out}`, /deadlock/); + assert.equal(deleted.out.trim(), '1', 'the delete took O with S'); + assert.equal(count(`id = ${q(dS)}`, 'rate.extension_submitters'), 0); + + // 10. The tables: anon and authenticated reach nothing; rate_worker reads and sets status only + // and has nothing on the key; service_role writes directly within the CHECKs. + const tables = [ + 'rate.extension_ingest_secret', + 'rate.extension_submitters', + 'rate.extension_observations', + ]; + for (const role of ['anon', 'authenticated']) { + for (const table of tables) { + refused( + role, + `SELECT count(*) FROM ${table}`, + '42501', + undefined, + `${role} select ${table}` + ); + refused( + role, + `DELETE FROM ${table}`, + '42501', + undefined, + `${role} delete ${table}` + ); + refused( + role, + `TRUNCATE ${table} CASCADE`, + '42501', + undefined, + `${role} truncate ${table}` + ); + } + refused( + role, + `UPDATE rate.extension_submitters SET created_at = now()`, + '42501', + undefined, + `${role} update submitters` + ); + refused( + role, + `UPDATE rate.extension_observations SET status = 'rejected'`, + '42501', + undefined, + `${role} update observations` + ); + refused( + role, + `UPDATE rate.extension_ingest_secret SET updated_at = now()`, + '42501', + undefined, + `${role} update the key` + ); + refused( + role, + 'INSERT INTO rate.extension_submitters DEFAULT VALUES', + '42501', + undefined, + `${role} insert submitter` + ); + refused( + role, + `INSERT INTO rate.extension_ingest_secret (key_sha256) VALUES (${q(sha('x'))})`, + '42501', + undefined, + `${role} insert key` + ); + refused( + role, + `INSERT INTO rate.extension_observations (submitter_id, schema_version, + marketplace, listing_url, external_item_id, price_vnd, stock_state, observed_at, extractor) + VALUES (${q(s1)}, 1, 'shopee', 'https://shopee.vn/x', 'x', 1000, 'unknown', now(), + '{"source": "dom", "extension_version": "1"}')`, + '42501', + undefined, + `${role} insert observation` + ); + } + refused( + 'rate_worker', + 'SELECT count(*) FROM rate.extension_ingest_secret', + '42501', + undefined, + 'rate_worker reads no key' + ); + assert.ok( + Number( + as('rate_worker', 'SELECT count(*) FROM rate.extension_observations') + ) > 0 + ); + assert.ok( + Number( + as('rate_worker', 'SELECT count(*) FROM rate.extension_submitters') + ) > 0 + ); + assert.equal( + as( + 'rate_worker', + `UPDATE rate.extension_observations SET status = 'corroborated' + WHERE id = ${r1.id} RETURNING status` + ), + 'corroborated' + ); + refused( + 'rate_worker', + `UPDATE rate.extension_observations SET status = 'promoted' + WHERE id = ${r1.id}`, + '23514', + undefined, + 'status is checked' + ); + refused( + 'rate_worker', + `UPDATE rate.extension_observations SET price_vnd = 1000 + WHERE id = ${r1.id}`, + '42501', + undefined, + 'rate_worker price' + ); + refused( + 'rate_worker', + `UPDATE rate.extension_observations SET network_hash = NULL + WHERE id = ${r1.id}`, + '42501', + undefined, + 'rate_worker hash' + ); + refused( + 'rate_worker', + `UPDATE rate.extension_submitters SET created_at = now()`, + '42501', + undefined, + 'rate_worker submitter' + ); + refused( + 'rate_worker', + `DELETE FROM rate.extension_observations WHERE id = ${r1.id}`, + '42501', + undefined, + 'rate_worker delete' + ); + refused( + 'rate_worker', + 'INSERT INTO rate.extension_submitters DEFAULT VALUES', + '42501', + undefined, + 'rate_worker insert' + ); + const direct = `INSERT INTO rate.extension_observations (submitter_id, schema_version, + marketplace, listing_url, external_item_id, price_vnd, stock_state, observed_at, extractor) + VALUES (${q(s1)}, 1, 'shopee', URL, 'x', 1000, 'unknown', now(), + '{"source": "dom", "extension_version": "1"}') RETURNING status`; + assert.equal( + as('service_role', direct.replace('URL', q('https://shopee.vn/x'))), + 'quarantined', + 'service_role inserts directly' + ); + for (const url of [ + 'https://lazada.vn/x', + 'https://shopee.vn/x;a=1', + `https://shopee.vn/x${ch(0xa0)}`, + ]) + refused( + 'service_role', + direct.replace('URL', q(url)), + '23514', + undefined, + `the URL CHECK holds service_role: ${url}` + ); + refused( + 'service_role', + `INSERT INTO rate.extension_submitters (network_hash) VALUES ('x')`, + '23514', + undefined, + 'the submitter hash CHECK' + ); + checkAll('after the exercise'); + + // 11. The rollback twice; everything goes; the proposal again needs the key row again. + sql(rollback); + sql(rollback); + assert.equal( + sql(`SELECT count(*) FROM pg_class WHERE relnamespace = 'rate'::regnamespace + AND relname LIKE 'extension\\_%'`), + '0' + ); + assert.equal( + sql(`SELECT count(*) FROM pg_proc WHERE pronamespace = 'rate'::regnamespace + AND proname IN ('register_extension_submitter', 'submit_extension_observation', + 'delete_extension_submitter', 'purge_extension_observations')`), + '0' + ); + sql(proposal); + assert.equal(sql(check('V12')), '1', 'V12 sees the missing key row'); + refused( + 'anon', + registerSql(NET_A), + '42501', + 'ingest_key', + 'no key row after a re-apply' + ); + sql( + `INSERT INTO rate.extension_ingest_secret (key_sha256) VALUES (${q(sha(KEY))})` + ); + checkAll('after the rollback and a re-apply'); + const again = register(); + assert.equal(submit(again, good()).status, 'quarantined'); + sql(proposal); + checkAll('after a re-apply over data'); + assert.equal(count(), 1, 'a re-apply keeps rows'); + assert.equal( + count('true', 'rate.extension_ingest_secret'), + 1, + 'a re-apply keeps the key' + ); + + console.log( + `extension observations proposal: ${valid.length} valid and ${bad.length} refused payloads agree with parseExtensionObservationV1; ingest key, exact limits, delete, purge and its race, grants, verify and rollback replayed` + ); +} finally { + removeContainer(); +}