laantungir/c-relay-pg
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v2.1.15 - Fix PG addressable d-tag derivation and replaceable/addressable transactional replace semantics

Browse Source
This commit is contained in:
Laan Tungir
2026-05-01 19:01:06 -04:00
parent 67cf3973c5
commit 8fdc362a8b
5 changed files with 183 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
relay.pid
+1 -1
View File
@@ -1 +1 @@
236636
257278
src/db_ops_postgres.c
+98 -30
View File
@@ -1620,37 +1620,82 @@ int postgres_db_insert_event_with_json(const char* id, const char* pubkey, long
id, pubkey, created_buf, kind_buf, event_type, content, sig, tags_json, event_json
};
PGresult* res = NULL;
if (strcmp(event_type, "replaceable") == 0) {
res = PQexecParams(conn,
"INSERT INTO events (id, pubkey, created_at, kind, event_type, content, sig, tags, event_json) "
"VALUES ($1, $2, $3::BIGINT, $4::INT, $5, $6, $7, $8::jsonb, $9) "
"ON CONFLICT (pubkey, kind) WHERE event_type = 'replaceable' DO UPDATE SET "
"id = EXCLUDED.id, created_at = EXCLUDED.created_at, content = EXCLUDED.content, "
"sig = EXCLUDED.sig, tags = EXCLUDED.tags, event_json = EXCLUDED.event_json "
"WHERE EXCLUDED.created_at > events.created_at "
"OR (EXCLUDED.created_at = events.created_at AND EXCLUDED.id < events.id)",
9, NULL, params, NULL, NULL, 0);
} else if (strcmp(event_type, "addressable") == 0) {
res = PQexecParams(conn,
"INSERT INTO events (id, pubkey, created_at, kind, event_type, content, sig, tags, event_json) "
"VALUES ($1, $2, $3::BIGINT, $4::INT, $5, $6, $7, $8::jsonb, $9) "
"ON CONFLICT (pubkey, kind, d_tag_value) WHERE event_type = 'addressable' DO UPDATE SET "
"id = EXCLUDED.id, created_at = EXCLUDED.created_at, content = EXCLUDED.content, "
"sig = EXCLUDED.sig, tags = EXCLUDED.tags, event_json = EXCLUDED.event_json "
"WHERE EXCLUDED.created_at > events.created_at "
"OR (EXCLUDED.created_at = events.created_at AND EXCLUDED.id < events.id)",
9, NULL, params, NULL, NULL, 0);
} else {
res = PQexecParams(conn,
"INSERT INTO events (id, pubkey, created_at, kind, event_type, content, sig, tags, event_json) "
"VALUES ($1, $2, $3::BIGINT, $4::INT, $5, $6, $7, $8::jsonb, $9) "
"ON CONFLICT (id) DO NOTHING",
9, NULL, params, NULL, NULL, 0);
int use_tx = (strcmp(event_type, "replaceable") == 0 || strcmp(event_type, "addressable") == 0);
if (use_tx) {
PGresult* tx_begin = PQexec(conn, "BEGIN");
if (!tx_begin || PQresultStatus(tx_begin) != PGRES_COMMAND_OK) {
if (tx_begin) {
postgres_set_error_text(PQresultErrorMessage(tx_begin));
PQclear(tx_begin);
} else {
postgres_set_error_from_conn(conn, "postgres_db_insert_event_with_json BEGIN failed");
}
if (out_step_rc) *out_step_rc = DB_ERROR;
if (out_extended_errcode) *out_extended_errcode = DB_ERROR;
return DB_ERROR;
}
PQclear(tx_begin);
PGresult* del_res = NULL;
if (strcmp(event_type, "replaceable") == 0) {
const char* del_params[4] = { pubkey, kind_buf, created_buf, id };
del_res = PQexecParams(conn,
"DELETE FROM events "
"WHERE event_type = 'replaceable' "
" AND pubkey = $1 "
" AND kind = $2::INT "
" AND (created_at < $3::BIGINT "
" OR (created_at = $3::BIGINT AND id > $4))",
4, NULL, del_params, NULL, NULL, 0);
} else {
const char* del_params[5] = { pubkey, kind_buf, created_buf, id, tags_json };
del_res = PQexecParams(conn,
"DELETE FROM events "
"WHERE event_type = 'addressable' "
" AND pubkey = $1 "
" AND kind = $2::INT "
" AND d_tag_value = COALESCE(( "
" SELECT tag->>1 "
" FROM jsonb_array_elements($5::jsonb) AS tag "
" WHERE jsonb_typeof(tag) = 'array' "
" AND jsonb_array_length(tag) >= 2 "
" AND tag->>0 = 'd' "
" LIMIT 1 "
" ), '') "
" AND (created_at < $3::BIGINT "
" OR (created_at = $3::BIGINT AND id > $4))",
5, NULL, del_params, NULL, NULL, 0);
}
if (!del_res || PQresultStatus(del_res) != PGRES_COMMAND_OK) {
if (del_res) {
postgres_set_error_text(PQresultErrorMessage(del_res));
PQclear(del_res);
} else {
postgres_set_error_from_conn(conn, "postgres_db_insert_event_with_json DELETE failed");
}
PGresult* tx_rollback = PQexec(conn, "ROLLBACK");
if (tx_rollback) PQclear(tx_rollback);
if (out_step_rc) *out_step_rc = DB_ERROR;
if (out_extended_errcode) *out_extended_errcode = DB_ERROR;
return DB_ERROR;
}
PQclear(del_res);
}
PGresult* res = PQexecParams(conn,
"INSERT INTO events (id, pubkey, created_at, kind, event_type, content, sig, tags, event_json) "
"VALUES ($1, $2, $3::BIGINT, $4::INT, $5, $6, $7, $8::jsonb, $9) "
"ON CONFLICT (id) DO NOTHING",
9, NULL, params, NULL, NULL, 0);
if (!res) {
postgres_set_error_from_conn(conn, "postgres_db_insert_event_with_json failed");
postgres_set_error_from_conn(conn, "postgres_db_insert_event_with_json INSERT failed");
if (use_tx) {
PGresult* tx_rollback = PQexec(conn, "ROLLBACK");
if (tx_rollback) PQclear(tx_rollback);
}
if (out_step_rc) *out_step_rc = DB_ERROR;
if (out_extended_errcode) *out_extended_errcode = DB_ERROR;
return DB_ERROR;
@@ -1668,15 +1713,38 @@ int postgres_db_insert_event_with_json(const char* id, const char* pubkey, long
}
postgres_set_error_text(PQresultErrorMessage(res));
PQclear(res);
if (use_tx) {
PGresult* tx_rollback = PQexec(conn, "ROLLBACK");
if (tx_rollback) PQclear(tx_rollback);
}
return is_constraint ? DB_OK : DB_ERROR;
}
const char* tuples = PQcmdTuples(res);
int inserted = (tuples && tuples[0] != '\0') ? atoi(tuples) : 0;
PQclear(res);
if (use_tx) {
PGresult* tx_commit = PQexec(conn, "COMMIT");
if (!tx_commit || PQresultStatus(tx_commit) != PGRES_COMMAND_OK) {
if (tx_commit) {
postgres_set_error_text(PQresultErrorMessage(tx_commit));
PQclear(tx_commit);
} else {
postgres_set_error_from_conn(conn, "postgres_db_insert_event_with_json COMMIT failed");
}
PGresult* tx_rollback = PQexec(conn, "ROLLBACK");
if (tx_rollback) PQclear(tx_rollback);
if (out_step_rc) *out_step_rc = DB_ERROR;
if (out_extended_errcode) *out_extended_errcode = DB_ERROR;
return DB_ERROR;
}
PQclear(tx_commit);
}
if (out_step_rc) *out_step_rc = inserted > 0 ? DB_DONE : DB_CONSTRAINT;
if (out_extended_errcode) *out_extended_errcode = inserted > 0 ? DB_OK : DB_CONSTRAINT;
PQclear(res);
return DB_OK;
#else
(void)id; (void)pubkey; (void)created_at; (void)kind; (void)event_type;
src/main.h
+2 -2
View File
@@ -13,8 +13,8 @@
// Using CRELAY_ prefix to avoid conflicts with nostr_core_lib VERSION macros
#define CRELAY_VERSION_MAJOR 2
#define CRELAY_VERSION_MINOR 1
#define CRELAY_VERSION_PATCH 14
#define CRELAY_VERSION "v2.1.14"
#define CRELAY_VERSION_PATCH 15
#define CRELAY_VERSION "v2.1.15"
// Relay metadata (authoritative source for NIP-11 information)
#define RELAY_NAME "C-Relay-PG"
src/pg_schema.h
+41 -20
View File
@@ -24,15 +24,7 @@ static const char* const EMBEDDED_PG_SCHEMA_SQL =
" content TEXT NOT NULL,\n"
" sig TEXT NOT NULL,\n"
" tags JSONB NOT NULL DEFAULT '[]'::jsonb,\n"
" d_tag_value TEXT GENERATED ALWAYS AS (\n"
" COALESCE(\n"
" NULLIF(\n"
" jsonb_path_query_first(tags, '$[*] ? (@[0] == \"d\")[1]') #>> '{}',\n"
" ''\n"
" ),\n"
" ''\n"
" )\n"
" ) STORED,\n"
" d_tag_value TEXT NOT NULL DEFAULT '',\n"
" expires_at BIGINT,\n"
" event_json TEXT NOT NULL,\n"
" first_seen BIGINT NOT NULL DEFAULT EXTRACT(EPOCH FROM NOW())::BIGINT\n"
@@ -43,24 +35,38 @@ static const char* const EMBEDDED_PG_SCHEMA_SQL =
"\n"
"DO $$\n"
"BEGIN\n"
" IF NOT EXISTS (\n"
" IF EXISTS (\n"
" SELECT 1\n"
" FROM information_schema.columns\n"
" WHERE table_schema = current_schema()\n"
" AND table_name = 'events'\n"
" AND column_name = 'd_tag_value'\n"
" AND is_generated = 'ALWAYS'\n"
" ) THEN\n"
" DROP INDEX IF EXISTS uq_events_addressable_pubkey_kind_dtag;\n"
" DROP INDEX IF EXISTS idx_events_d_tag_value;\n"
" ALTER TABLE events DROP COLUMN d_tag_value;\n"
" ALTER TABLE events ADD COLUMN d_tag_value TEXT NOT NULL DEFAULT '';\n"
" ELSIF NOT EXISTS (\n"
" SELECT 1\n"
" FROM information_schema.columns\n"
" WHERE table_schema = current_schema()\n"
" AND table_name = 'events'\n"
" AND column_name = 'd_tag_value'\n"
" ) THEN\n"
" ALTER TABLE events\n"
" ADD COLUMN d_tag_value TEXT GENERATED ALWAYS AS (\n"
" COALESCE(\n"
" NULLIF(\n"
" jsonb_path_query_first(tags, '$[*] ? (@[0] == \"d\")[1]') #>> '{}',\n"
" ''\n"
" ),\n"
" ''\n"
" )\n"
" ) STORED;\n"
" ALTER TABLE events ADD COLUMN d_tag_value TEXT NOT NULL DEFAULT '';\n"
" END IF;\n"
"\n"
" UPDATE events e\n"
" SET d_tag_value = COALESCE((\n"
" SELECT tag->>1\n"
" FROM jsonb_array_elements(e.tags) AS tag\n"
" WHERE jsonb_typeof(tag) = 'array'\n"
" AND jsonb_array_length(tag) >= 2\n"
" AND tag->>0 = 'd'\n"
" LIMIT 1\n"
" ), '')\n"
" WHERE COALESCE(e.d_tag_value, '') = '';\n"
"END\n"
"$$;\n"
"\n"
@@ -98,8 +104,10 @@ static const char* const EMBEDDED_PG_SCHEMA_SQL =
"RETURNS TRIGGER AS $$\n"
"DECLARE\n"
" expiration_value TEXT;\n"
" d_value TEXT;\n"
"BEGIN\n"
" NEW.expires_at := NULL;\n"
" NEW.d_tag_value := '';\n"
"\n"
" SELECT tag->>1\n"
" INTO expiration_value\n"
@@ -113,6 +121,19 @@ static const char* const EMBEDDED_PG_SCHEMA_SQL =
" NEW.expires_at := expiration_value::BIGINT;\n"
" END IF;\n"
"\n"
"\n"
" SELECT tag->>1\n"
" INTO d_value\n"
" FROM jsonb_array_elements(NEW.tags) AS tag\n"
" WHERE jsonb_typeof(tag) = 'array'\n"
" AND jsonb_array_length(tag) >= 2\n"
" AND tag->>0 = 'd'\n"
" LIMIT 1;\n"
"\n"
" IF d_value IS NOT NULL THEN\n"
" NEW.d_tag_value := d_value;\n"
" END IF;\n"
"\n"
" RETURN NEW;\n"
"END;\n"
"$$ LANGUAGE plpgsql;\n"
src/pg_schema.sql
+41 -21
View File
@@ -18,15 +18,7 @@ CREATE TABLE IF NOT EXISTS events (
content TEXT NOT NULL,
sig TEXT NOT NULL,
tags JSONB NOT NULL DEFAULT '[]'::jsonb,
d_tag_value TEXT GENERATED ALWAYS AS (
COALESCE(
NULLIF(
jsonb_path_query_first(tags, '$[*] ? (@[0] == "d")[1]') #>> '{}',
''
),
''
)
) STORED,
d_tag_value TEXT NOT NULL DEFAULT '',
expires_at BIGINT,
event_json TEXT NOT NULL,
first_seen BIGINT NOT NULL DEFAULT EXTRACT(EPOCH FROM NOW())::BIGINT
@@ -37,24 +29,38 @@ ALTER TABLE events
DO $$
BEGIN
IF NOT EXISTS (
IF EXISTS (
SELECT 1
FROM information_schema.columns
WHERE table_schema = current_schema()
AND table_name = 'events'
AND column_name = 'd_tag_value'
AND is_generated = 'ALWAYS'
) THEN
DROP INDEX IF EXISTS uq_events_addressable_pubkey_kind_dtag;
DROP INDEX IF EXISTS idx_events_d_tag_value;
ALTER TABLE events DROP COLUMN d_tag_value;
ALTER TABLE events ADD COLUMN d_tag_value TEXT NOT NULL DEFAULT '';
ELSIF NOT EXISTS (
SELECT 1
FROM information_schema.columns
WHERE table_schema = current_schema()
AND table_name = 'events'
AND column_name = 'd_tag_value'
) THEN
ALTER TABLE events
ADD COLUMN d_tag_value TEXT GENERATED ALWAYS AS (
COALESCE(
NULLIF(
jsonb_path_query_first(tags, '$[*] ? (@[0] == "d")[1]') #>> '{}',
''
),
''
)
) STORED;
ALTER TABLE events ADD COLUMN d_tag_value TEXT NOT NULL DEFAULT '';
END IF;
UPDATE events e
SET d_tag_value = COALESCE((
SELECT tag->>1
FROM jsonb_array_elements(e.tags) AS tag
WHERE jsonb_typeof(tag) = 'array'
AND jsonb_array_length(tag) >= 2
AND tag->>0 = 'd'
LIMIT 1
), '')
WHERE COALESCE(e.d_tag_value, '') = '';
END
$$;
@@ -92,8 +98,10 @@ CREATE OR REPLACE FUNCTION set_event_derived_fields()
RETURNS TRIGGER AS $$
DECLARE
expiration_value TEXT;
d_value TEXT;
BEGIN
NEW.expires_at := NULL;
NEW.d_tag_value := '';
SELECT tag->>1
INTO expiration_value
@@ -107,6 +115,18 @@ BEGIN
NEW.expires_at := expiration_value::BIGINT;
END IF;
SELECT tag->>1
INTO d_value
FROM jsonb_array_elements(NEW.tags) AS tag
WHERE jsonb_typeof(tag) = 'array'
AND jsonb_array_length(tag) >= 2
AND tag->>0 = 'd'
LIMIT 1;
IF d_value IS NOT NULL THEN
NEW.d_tag_value := d_value;
END IF;
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
@@ -232,7 +252,7 @@ CREATE TABLE IF NOT EXISTS ip_bans (
);
INSERT INTO schema_info(key, value, updated_at)
VALUES ('version', '2', EXTRACT(EPOCH FROM NOW())::BIGINT)
VALUES ('version', '3', EXTRACT(EPOCH FROM NOW())::BIGINT)
ON CONFLICT (key) DO UPDATE SET
value = EXCLUDED.value,
updated_at = EXCLUDED.updated_at;