sql_postgresql.c

//
//  sql_postgresql.c
//  cloudsync
//
//  PostgreSQL-specific SQL queries
//  Created by Claude Code on 22/12/25.
//

#include "../sql.h"

// MARK: Settings

const char * const SQL_SETTINGS_GET_VALUE =
    "SELECT value FROM cloudsync_settings WHERE key=$1;";

const char * const SQL_SETTINGS_SET_KEY_VALUE_REPLACE =
    "INSERT INTO cloudsync_settings (key, value) VALUES ($1, $2) "
    "ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value;";

const char * const SQL_SETTINGS_SET_KEY_VALUE_DELETE =
    "DELETE FROM cloudsync_settings WHERE key = $1;";

const char * const SQL_TABLE_SETTINGS_GET_VALUE =
    "SELECT value FROM cloudsync_table_settings WHERE (tbl_name=$1 AND col_name=$2 AND key=$3);";

const char * const SQL_TABLE_SETTINGS_DELETE_ALL_FOR_TABLE =
    "DELETE FROM cloudsync_table_settings WHERE tbl_name=$1;";

const char * const SQL_TABLE_SETTINGS_REPLACE =
    "INSERT INTO cloudsync_table_settings (tbl_name, col_name, key, value) VALUES ($1, $2, $3, $4) "
    "ON CONFLICT (tbl_name, key) DO UPDATE SET col_name = EXCLUDED.col_name, value = EXCLUDED.value;";

const char * const SQL_TABLE_SETTINGS_DELETE_ONE =
    "DELETE FROM cloudsync_table_settings WHERE (tbl_name=$1 AND col_name=$2 AND key=$3);";

const char * const SQL_TABLE_SETTINGS_COUNT_TABLES =
    "SELECT count(*) FROM cloudsync_table_settings WHERE key='algo';";

const char * const SQL_SETTINGS_LOAD_GLOBAL =
    "SELECT key, value FROM cloudsync_settings;";

const char * const SQL_SETTINGS_LOAD_TABLE =
    "SELECT lower(tbl_name), lower(col_name), key, value FROM cloudsync_table_settings ORDER BY tbl_name;";

const char * const SQL_CREATE_SETTINGS_TABLE =
    "CREATE TABLE IF NOT EXISTS cloudsync_settings (key TEXT PRIMARY KEY NOT NULL, value TEXT);"
    "CREATE TABLE IF NOT EXISTS public.app_schema_version ("
    "version BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY"
    ");"
    "CREATE OR REPLACE FUNCTION bump_app_schema_version() "
    "RETURNS event_trigger AS $$ "
    "BEGIN "
    "INSERT INTO public.app_schema_version DEFAULT VALUES; "
    "END;"
    "$$ LANGUAGE plpgsql;"
    "DROP EVENT TRIGGER IF EXISTS app_schema_change;"
    "CREATE EVENT TRIGGER app_schema_change "
    "ON ddl_command_end "
    "EXECUTE FUNCTION bump_app_schema_version();";

// format strings (snprintf) are also static SQL templates
const char * const SQL_INSERT_SETTINGS_STR_FORMAT =
    "INSERT INTO cloudsync_settings (key, value) VALUES ('%s', '%s');";

const char * const SQL_INSERT_SETTINGS_INT_FORMAT =
    "INSERT INTO cloudsync_settings (key, value) VALUES ('%s', %lld);";

const char * const SQL_CREATE_SITE_ID_TABLE =
    "CREATE TABLE IF NOT EXISTS cloudsync_site_id ("
    "id BIGSERIAL PRIMARY KEY, "
    "site_id BYTEA UNIQUE NOT NULL"
    ");";

const char * const SQL_INSERT_SITE_ID_ROWID =
    "INSERT INTO cloudsync_site_id (id, site_id) VALUES ($1, $2);";

const char * const SQL_CREATE_TABLE_SETTINGS_TABLE =
    "CREATE TABLE IF NOT EXISTS cloudsync_table_settings (tbl_name TEXT NOT NULL, col_name TEXT NOT NULL, key TEXT, value TEXT, PRIMARY KEY(tbl_name,key));";

const char * const SQL_CREATE_SCHEMA_VERSIONS_TABLE =
    "CREATE TABLE IF NOT EXISTS cloudsync_schema_versions (hash BIGINT PRIMARY KEY, seq INTEGER NOT NULL)";

const char * const SQL_SETTINGS_CLEANUP_DROP_ALL =
    "DROP TABLE IF EXISTS cloudsync_settings CASCADE; "
    "DROP TABLE IF EXISTS cloudsync_site_id CASCADE; "
    "DROP TABLE IF EXISTS cloudsync_table_settings CASCADE; "
    "DROP TABLE IF EXISTS cloudsync_schema_versions CASCADE;";

// MARK: CloudSync

const char * const SQL_DBVERSION_BUILD_QUERY =
    "WITH table_names AS ("
    "SELECT quote_ident(schemaname) || '.' || quote_ident(tablename) as tbl_name "
    "FROM pg_tables "
    "WHERE tablename LIKE '%_cloudsync'"
    "), "
    "query_parts AS ("
    "SELECT tbl_name, "
    "format('SELECT COALESCE(MAX(db_version), 0) FROM %s', tbl_name) as part "
    "FROM table_names"
    ") "
    "SELECT string_agg(part, ' UNION ALL ') FROM query_parts;";

const char * const SQL_CHANGES_INSERT_ROW =
    "INSERT INTO cloudsync_changes(tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq) "
    "VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9);";

// MARK: Additional SQL constants for PostgreSQL

const char * const SQL_SITEID_SELECT_ROWID0 =
    "SELECT site_id FROM cloudsync_site_id WHERE id = 0;";

const char * const SQL_DATA_VERSION =
    "SELECT txid_snapshot_xmin(txid_current_snapshot());";  // was "PRAGMA data_version"

const char * const SQL_SCHEMA_VERSION =
    "SELECT COALESCE(max(version), 0) FROM app_schema_version;";  // was "PRAGMA schema_version"

const char * const SQL_SITEID_GETSET_ROWID_BY_SITEID =
    "INSERT INTO cloudsync_site_id (site_id) VALUES ($1) "
    "ON CONFLICT(site_id) DO UPDATE SET site_id = EXCLUDED.site_id "
    "RETURNING id;";

const char * const SQL_BUILD_SELECT_NONPK_COLS_BY_ROWID =
    "SELECT string_agg(quote_ident(column_name), ',' ORDER BY ordinal_position) "
    "FROM information_schema.columns "
    "WHERE table_name = $1 AND column_name NOT IN ("
    "SELECT column_name FROM information_schema.key_column_usage "
    "WHERE table_name = $1 AND constraint_name LIKE '%_pkey'"
    ");";  // TODO: build full SELECT ... WHERE ctid=? analog with ordered columns like SQLite

const char * const SQL_BUILD_SELECT_NONPK_COLS_BY_PK =
    "WITH tbl AS ("
    "  SELECT to_regclass('%s') AS oid"
    "), "
    "pk AS ("
    "  SELECT a.attname, k.ord "
    "  FROM pg_index x "
    "  JOIN tbl t ON t.oid = x.indrelid "
    "  JOIN LATERAL unnest(x.indkey) WITH ORDINALITY AS k(attnum, ord) ON true "
    "  JOIN pg_attribute a ON a.attrelid = x.indrelid AND a.attnum = k.attnum "
    "  WHERE x.indisprimary "
    "  ORDER BY k.ord"
    "), "
    "nonpk AS ("
    "  SELECT a.attname "
    "  FROM pg_attribute a "
    "  JOIN tbl t ON t.oid = a.attrelid "
    "  WHERE a.attnum > 0 AND NOT a.attisdropped "
    "    AND a.attnum NOT IN ("
    "      SELECT k.attnum "
    "      FROM pg_index x "
    "      JOIN tbl t2 ON t2.oid = x.indrelid "
    "      JOIN LATERAL unnest(x.indkey) AS k(attnum) ON true "
    "      WHERE x.indisprimary"
    "    ) "
    "  ORDER BY a.attnum"
    ") "
    "SELECT "
    "  'SELECT '"
    "  || (SELECT string_agg(format('%%I', attname), ',') FROM nonpk)"
    "  || ' FROM ' || (SELECT (oid::regclass)::text FROM tbl)"
    "  || ' WHERE '"
    "  || (SELECT string_agg(format('%%I=$%%s', attname, ord), ' AND ' ORDER BY ord) FROM pk)"
    "  || ';';";

const char * const SQL_DELETE_ROW_BY_ROWID =
    "DELETE FROM %s WHERE ctid = $1;";  // TODO: consider using PK-based deletion; ctid is unstable

const char * const SQL_BUILD_DELETE_ROW_BY_PK =
    "WITH tbl AS ("
    "  SELECT to_regclass('%s') AS oid"
    "), "
    "pk AS ("
    "  SELECT a.attname, k.ord, format_type(a.atttypid, a.atttypmod) AS coltype "
    "  FROM pg_index x "
    "  JOIN tbl t ON t.oid = x.indrelid "
    "  JOIN LATERAL unnest(x.indkey) WITH ORDINALITY AS k(attnum, ord) ON true "
    "  JOIN pg_attribute a ON a.attrelid = x.indrelid AND a.attnum = k.attnum "
    "  WHERE x.indisprimary "
    "  ORDER BY k.ord"
    ") "
    "SELECT "
    "  'DELETE FROM ' || (SELECT (oid::regclass)::text FROM tbl)"
    "  || ' WHERE '"
    "  || (SELECT string_agg(format('%%I=$%%s::%%s', attname, ord, coltype), ' AND ' ORDER BY ord) FROM pk)"
    "  || ';';";

const char * const SQL_INSERT_ROWID_IGNORE =
    "INSERT INTO %s DEFAULT VALUES ON CONFLICT DO NOTHING;";  // TODO: adapt to explicit PK inserts (no rowid in PG)

const char * const SQL_UPSERT_ROWID_AND_COL_BY_ROWID =
    "INSERT INTO %s (ctid, %s) VALUES ($1, $2) "
    "ON CONFLICT DO UPDATE SET %s = $2;";  // TODO: align with SQLite upsert by rowid; avoid ctid

const char * const SQL_BUILD_INSERT_PK_IGNORE =
    "WITH tbl AS ("
    "  SELECT to_regclass('%s') AS oid"
    "), "
    "pk AS ("
    "  SELECT a.attname, k.ord, format_type(a.atttypid, a.atttypmod) AS coltype "
    "  FROM pg_index x "
    "  JOIN tbl t ON t.oid = x.indrelid "
    "  JOIN LATERAL unnest(x.indkey) WITH ORDINALITY AS k(attnum, ord) ON true "
    "  JOIN pg_attribute a ON a.attrelid = x.indrelid AND a.attnum = k.attnum "
    "  WHERE x.indisprimary "
    "  ORDER BY k.ord"
    ") "
    "SELECT "
    "  'INSERT INTO ' || (SELECT (oid::regclass)::text FROM tbl)"
    "  || ' (' || (SELECT string_agg(format('%%I', attname), ',') FROM pk) || ')'"
    "  || ' VALUES (' || (SELECT string_agg(format('$%%s::%%s', ord, coltype), ',') FROM pk) || ')'"
    "  || ' ON CONFLICT DO NOTHING;';";

const char * const SQL_BUILD_UPSERT_PK_AND_COL =
    "WITH tbl AS ("
    "  SELECT to_regclass('%s') AS oid"
    "), "
    "pk AS ("
    "  SELECT a.attname, k.ord, format_type(a.atttypid, a.atttypmod) AS coltype "
    "  FROM pg_index x "
    "  JOIN tbl t ON t.oid = x.indrelid "
    "  JOIN LATERAL unnest(x.indkey) WITH ORDINALITY AS k(attnum, ord) ON true "
    "  JOIN pg_attribute a ON a.attrelid = x.indrelid AND a.attnum = k.attnum "
    "  WHERE x.indisprimary "
    "  ORDER BY k.ord"
    "), "
    "pk_count AS ("
    "  SELECT count(*) AS n FROM pk"
    "), "
    "col AS ("
    "  SELECT '%s'::text AS colname"
    ") "
    "SELECT "
    "  'INSERT INTO ' || (SELECT (oid::regclass)::text FROM tbl)"
    "  || ' (' || (SELECT string_agg(format('%%I', attname), ',') FROM pk)"
    "  || ',' || (SELECT format('%%I', colname) FROM col) || ')'"
    "  || ' VALUES (' || (SELECT string_agg(format('$%%s::%%s', ord, coltype), ',') FROM pk)"
    "  || ',' || (SELECT format('$%%s', (SELECT n FROM pk_count) + 1)) || ')'"
    "  || ' ON CONFLICT (' || (SELECT string_agg(format('%%I', attname), ',') FROM pk) || ')'"
    "  || ' DO UPDATE SET ' || (SELECT format('%%I', colname) FROM col)"
    "  || '=' || (SELECT format('$%%s', (SELECT n FROM pk_count) + 2)) || ';';";

const char * const SQL_SELECT_COLS_BY_ROWID_FMT =
    "SELECT %s%s%s FROM %s WHERE ctid = $1;";  // TODO: align with PK/rowid selection builder

const char * const SQL_BUILD_SELECT_COLS_BY_PK_FMT =
    "WITH tbl AS ("
    "  SELECT to_regclass('%s') AS tblreg"
    "), "
    "pk AS ("
    "  SELECT a.attname, k.ord, format_type(a.atttypid, a.atttypmod) AS coltype "
    "  FROM pg_index x "
    "  JOIN tbl t ON t.tblreg = x.indrelid "
    "  JOIN LATERAL unnest(x.indkey) WITH ORDINALITY AS k(attnum, ord) ON true "
    "  JOIN pg_attribute a ON a.attrelid = x.indrelid AND a.attnum = k.attnum "
    "  WHERE x.indisprimary "
    "  ORDER BY k.ord"
    "), "
    "col AS ("
    "  SELECT '%s'::text AS colname"
    ") "
    "SELECT "
    "  'SELECT ' || (SELECT format('%%I', colname) FROM col) "
    "  || ' FROM ' || (SELECT tblreg::text FROM tbl)"
    "  || ' WHERE '"
    "  || (SELECT string_agg(format('%%I=$%%s::%%s', attname, ord, coltype), ' AND ' ORDER BY ord) FROM pk)"
    "  || ';';";

const char * const SQL_CLOUDSYNC_ROW_EXISTS_BY_PK =
    "SELECT EXISTS(SELECT 1 FROM %s WHERE pk = $1 LIMIT 1);";

const char * const SQL_CLOUDSYNC_UPDATE_COL_BUMP_VERSION =
    "UPDATE %s "
    "SET col_version = CASE col_version %% 2 WHEN 0 THEN col_version + 1 ELSE col_version + 2 END, "
    "db_version = $1, seq = $2, site_id = 0 "
    "WHERE pk = $3 AND col_name = '%s';";

const char * const SQL_CLOUDSYNC_UPSERT_COL_INIT_OR_BUMP_VERSION =
    "INSERT INTO %s (pk, col_name, col_version, db_version, seq, site_id) "
    "VALUES ($1, '%s', 1, $2, $3, 0) "
    "ON CONFLICT (pk, col_name) DO UPDATE SET "
    "col_version = CASE EXCLUDED.col_version %% 2 WHEN 0 THEN EXCLUDED.col_version + 1 ELSE EXCLUDED.col_version + 2 END, "
    "db_version = $2, seq = $3, site_id = 0;";  // TODO: mirror SQLite's bump rules and bind usage

const char * const SQL_CLOUDSYNC_UPSERT_RAW_COLVERSION =
    "INSERT INTO %s (pk, col_name, col_version, db_version, seq, site_id) "
    "VALUES ($1, $2, $3, $4, $5, 0) "
    "ON CONFLICT (pk, col_name) DO UPDATE SET "
    "col_version = %s.col_version + 1, db_version = $6, seq = $7, site_id = 0;";

const char * const SQL_CLOUDSYNC_DELETE_PK_EXCEPT_COL =
    "DELETE FROM %s WHERE pk = $1 AND col_name != '%s';";  // TODO: match SQLite delete semantics

const char * const SQL_CLOUDSYNC_REKEY_PK_AND_RESET_VERSION_EXCEPT_COL =
    "WITH moved AS ("
    "  SELECT col_name "
    "  FROM %s WHERE pk = $3 AND col_name != '%s'"
    "), "
    "upserted AS ("
    "  INSERT INTO %s (pk, col_name, col_version, db_version, seq, site_id) "
    "  SELECT $1, col_name, 1, $2, cloudsync_seq(), 0 "
    "  FROM moved "
    "  ON CONFLICT (pk, col_name) DO UPDATE SET "
    "  col_version = 1, db_version = $2, seq = cloudsync_seq(), site_id = 0"
    ") "
    "DELETE FROM %s WHERE pk = $3 AND col_name != '%s';";

const char * const SQL_CLOUDSYNC_GET_COL_VERSION_OR_ROW_EXISTS =
    "SELECT COALESCE("
    "(SELECT col_version FROM %s WHERE pk = $1 AND col_name = '%s'), "
    "(SELECT 1 FROM %s WHERE pk = $1 LIMIT 1)"
    ");";

const char * const SQL_CLOUDSYNC_INSERT_RETURN_CHANGE_ID =
    "INSERT INTO %s "
    "(pk, col_name, col_version, db_version, seq, site_id) "
    "VALUES ($1, $2, $3, cloudsync_db_version_next($4), $5, $6) "
    "ON CONFLICT (pk, col_name) DO UPDATE SET "
    "col_version = EXCLUDED.col_version, "
    "db_version = cloudsync_db_version_next($4), "
    "seq = EXCLUDED.seq, "
    "site_id = EXCLUDED.site_id "
    "RETURNING ((db_version::bigint << 30) | seq);";  // TODO: align RETURNING and bump logic with SQLite (version increments on conflict)

const char * const SQL_CLOUDSYNC_TOMBSTONE_PK_EXCEPT_COL =
    "UPDATE %s "
    "SET col_version = 0, db_version = cloudsync_db_version_next($1) "
    "WHERE pk = $2 AND col_name != '%s';";  // TODO: confirm tombstone semantics match SQLite

const char * const SQL_CLOUDSYNC_SELECT_COL_VERSION_BY_PK_COL =
    "SELECT col_version FROM %s WHERE pk = $1 AND col_name = $2;";

const char * const SQL_CLOUDSYNC_SELECT_SITE_ID_BY_PK_COL =
    "SELECT site_id FROM %s WHERE pk = $1 AND col_name = $2;";

const char * const SQL_PRAGMA_TABLEINFO_LIST_NONPK_NAME_CID =
    "SELECT c.column_name, c.ordinal_position "
    "FROM information_schema.columns c "
    "WHERE c.table_name = '%s' "
    "AND c.table_schema = COALESCE(NULLIF('%s', ''), current_schema()) "
    "AND c.column_name NOT IN ("
    "  SELECT kcu.column_name FROM information_schema.table_constraints tc "
    "  JOIN information_schema.key_column_usage kcu ON tc.constraint_name = kcu.constraint_name "
    "    AND tc.table_schema = kcu.table_schema "
    "  WHERE tc.table_name = '%s' AND tc.table_schema = COALESCE(NULLIF('%s', ''), current_schema()) "
    "  AND tc.constraint_type = 'PRIMARY KEY'"
    ") "
    "ORDER BY ordinal_position;";

const char * const SQL_DROP_CLOUDSYNC_TABLE =
    "DROP TABLE IF EXISTS %s CASCADE;";

const char * const SQL_CLOUDSYNC_DELETE_COLS_NOT_IN_SCHEMA_OR_PKCOL =
    "DELETE FROM %s WHERE col_name NOT IN ("
    "SELECT column_name FROM information_schema.columns WHERE table_name = '%s' "
    "AND table_schema = COALESCE(NULLIF('%s', ''), current_schema()) "
    "UNION SELECT '%s'"
    ");";

const char * const SQL_PRAGMA_TABLEINFO_PK_QUALIFIED_COLLIST_FMT =
    "SELECT string_agg(quote_ident(column_name), ',' ORDER BY ordinal_position) "
    "FROM information_schema.key_column_usage "
    "WHERE table_name = '%s' AND table_schema = COALESCE(cloudsync_schema(), current_schema()) "
    "AND constraint_name LIKE '%%_pkey';";

const char * const SQL_CLOUDSYNC_GC_DELETE_ORPHANED_PK =
    "DELETE FROM %s "
    "WHERE (col_name != '%s' OR (col_name = '%s' AND col_version %% 2 != 0)) "
    "AND NOT EXISTS ("
    "SELECT 1 FROM %s "
    "WHERE %s.pk = cloudsync_pk_encode(%s) LIMIT 1"
    ");";

const char * const SQL_PRAGMA_TABLEINFO_PK_COLLIST =
    "SELECT string_agg(quote_ident(column_name), ',') "
    "FROM information_schema.key_column_usage "
    "WHERE table_name = '%s' AND table_schema = COALESCE(cloudsync_schema(), current_schema()) "
    "AND constraint_name LIKE '%%_pkey';";

const char * const SQL_PRAGMA_TABLEINFO_PK_DECODE_SELECTLIST =
    "SELECT string_agg("
    "'cloudsync_pk_decode(pk, ' || ordinal_position || ') AS ' || quote_ident(column_name), ',' ORDER BY ordinal_position"
    ") "
    "FROM information_schema.key_column_usage "
    "WHERE table_name = '%s' AND table_schema = COALESCE(cloudsync_schema(), current_schema()) "
    "AND constraint_name LIKE '%%_pkey';";

const char * const SQL_CLOUDSYNC_INSERT_MISSING_PKS_FROM_BASE_EXCEPT_SYNC =
    "SELECT cloudsync_insert('%s', %s) "
    "FROM (SELECT %s FROM %s EXCEPT SELECT %s FROM %s);";

const char * const SQL_CLOUDSYNC_SELECT_PKS_NOT_IN_SYNC_FOR_COL =
    "WITH _cstemp1 AS (SELECT cloudsync_pk_encode(%s) AS pk FROM %s) "
    "SELECT _cstemp1.pk FROM _cstemp1 "
    "WHERE NOT EXISTS ("
    "SELECT 1 FROM %s _cstemp2 "
    "WHERE _cstemp2.pk = _cstemp1.pk AND _cstemp2.col_name = $1"
    ");";