xiaoshu-app/scripts/sql/read-revisions.sql

BEGIN;
-- Append one event per writer transaction/domain. No shared counter-row locks.
CREATE TABLE IF NOT EXISTS xs_read_revision_event (
 company text NOT NULL,domain text NOT NULL,revision bigint NOT NULL,
 PRIMARY KEY(company,domain,revision)
);
CREATE TABLE IF NOT EXISTS xs_history_summary (company text NOT NULL,actor text NOT NULL,key text NOT NULL,revision text NOT NULL,value jsonb NOT NULL,expires_at timestamptz NOT NULL,PRIMARY KEY(company,actor,key));
CREATE OR REPLACE FUNCTION xs_read_version(co text,scope text) RETURNS text LANGUAGE sql STABLE AS $$
 WITH frontier AS (SELECT GREATEST(
  COALESCE((SELECT max(revision) FROM xs_read_revision_event WHERE company=co AND domain=scope),0),
  COALESCE((SELECT max(revision) FROM xs_read_revision_event WHERE company='*' AND domain=scope),0)) revision)
 SELECT revision::text||':'||COALESCE((SELECT string_agg(x::text,',' ORDER BY x)
  FROM txid_snapshot_xip(txid_current_snapshot()) x WHERE x<=frontier.revision),'') FROM frontier
$$;
-- Include older in-flight writer IDs in the version. If an older writer commits
-- after the frontier, its disappearance invalidates the cache (MAX alone cannot).
CREATE OR REPLACE FUNCTION xs_prune_read_revisions() RETURNS void LANGUAGE sql AS $$
 DELETE FROM xs_read_revision_event e USING (
  SELECT company,domain,max(revision) keep FROM xs_read_revision_event
  WHERE revision<txid_snapshot_xmin(txid_current_snapshot()) GROUP BY company,domain
 ) k WHERE e.company=k.company AND e.domain=k.domain AND e.revision<k.keep;
 DELETE FROM xs_history_summary WHERE expires_at<=now();
$$;
CREATE OR REPLACE FUNCTION xs_bump_read_revision() RETURNS trigger LANGUAGE plpgsql AS $$
DECLARE companies text[];co text;scope text;
BEGIN
 IF TG_OP='TRUNCATE' THEN
  INSERT INTO xs_read_revision_event(company,domain,revision) SELECT '*',domain,txid_current() FROM unnest(string_to_array(TG_ARGV[0],',')) domain ON CONFLICT DO NOTHING;
  RETURN NULL;
 ELSIF TG_OP='INSERT' THEN
  SELECT array_agg(DISTINCT COALESCE(to_jsonb(n)->>'company','*')) INTO companies FROM new_rows n;
 ELSIF TG_OP='DELETE' THEN
  SELECT array_agg(DISTINCT COALESCE(to_jsonb(o)->>'company','*')) INTO companies FROM old_rows o;
 ELSE
  SELECT array_agg(DISTINCT company ORDER BY company) INTO companies FROM (
   SELECT COALESCE(to_jsonb(n)->>'company','*') company FROM new_rows n
   UNION SELECT COALESCE(to_jsonb(o)->>'company','*') FROM old_rows o
  ) x;
 END IF;
 -- Keys are transaction-specific; different writers never contend on a counter.
 FOR co IN SELECT unnest(companies) ORDER BY 1 LOOP
  FOR scope IN SELECT unnest(string_to_array(TG_ARGV[0],',')) ORDER BY 1 LOOP
   INSERT INTO xs_read_revision_event(company,domain,revision) VALUES(co,scope,txid_current()) ON CONFLICT DO NOTHING;
  END LOOP;
 END LOOP;
 RETURN NULL;
END $$;
DO $$
DECLARE source record;
BEGIN
 FOR source IN SELECT * FROM (VALUES
 ('LessonRecord','account,lessons'),('CommonModel','learning,lessons'),
 ('LessonCompletionReport','lessons'),('CourseAppointment','appointments,lessons'),
 ('_User','account,lessons'),('Node','courses,lessons'),('CourseBinding','courses'),
 ('DailyStudyRecord','learning,lessons'),('MemoryPracticeRecord','lessons,review'),
 ('CourseCreditReservation','account'),('SurveyItem','reading'),('SurveyLog','reading')
 ) s(tbl,domains) LOOP
  IF to_regclass(format('%I',source.tbl)) IS NULL THEN CONTINUE;END IF;
  EXECUTE format('DROP TRIGGER IF EXISTS xs_revision_insert ON %I',source.tbl);
  EXECUTE format('DROP TRIGGER IF EXISTS xs_revision_update ON %I',source.tbl);
  EXECUTE format('DROP TRIGGER IF EXISTS xs_revision_delete ON %I',source.tbl);
  EXECUTE format('DROP TRIGGER IF EXISTS xs_revision_truncate ON %I',source.tbl);
  EXECUTE format('CREATE TRIGGER xs_revision_insert AFTER INSERT ON %I REFERENCING NEW TABLE AS new_rows FOR EACH STATEMENT EXECUTE FUNCTION xs_bump_read_revision(%L)',source.tbl,source.domains);
  EXECUTE format('CREATE TRIGGER xs_revision_update AFTER UPDATE ON %I REFERENCING OLD TABLE AS old_rows NEW TABLE AS new_rows FOR EACH STATEMENT EXECUTE FUNCTION xs_bump_read_revision(%L)',source.tbl,source.domains);
  EXECUTE format('CREATE TRIGGER xs_revision_delete AFTER DELETE ON %I REFERENCING OLD TABLE AS old_rows FOR EACH STATEMENT EXECUTE FUNCTION xs_bump_read_revision(%L)',source.tbl,source.domains);
  EXECUTE format('CREATE TRIGGER xs_revision_truncate AFTER TRUNCATE ON %I FOR EACH STATEMENT EXECUTE FUNCTION xs_bump_read_revision(%L)',source.tbl,source.domains);
 END LOOP;
END $$;
COMMIT;