diff --git a/backend/src/api/contracts.ts b/backend/src/api/contracts.ts index 45287b0..4832580 100644 --- a/backend/src/api/contracts.ts +++ b/backend/src/api/contracts.ts @@ -91,6 +91,34 @@ const FEED_MESSAGE_HISTORY_SCHEMA = { }, }; +const OWNER_PACKET_SHARING_SCHEMA = { + type: 'object', + required: ['nodeId', 'featureEnabled', 'enabled', 'destinations'], + properties: { + nodeId: { type: 'string', pattern: '^[0-9A-Fa-f]{64}$' }, + featureEnabled: { type: 'boolean' }, + enabled: { type: 'boolean' }, + destinations: { + type: 'array', + items: { + type: 'object', + required: ['id', 'name', 'websiteUrl', 'description', 'configured', 'selected', 'lastForwardedAt'], + properties: { + id: { type: 'string', pattern: '^[a-z0-9][a-z0-9_-]{0,63}$' }, + name: { type: 'string', minLength: 1, maxLength: 120 }, + websiteUrl: { type: ['string', 'null'], format: 'uri', maxLength: 2_048 }, + description: { type: ['string', 'null'], maxLength: 160 }, + configured: { type: 'boolean' }, + selected: { type: 'boolean' }, + lastForwardedAt: { type: ['string', 'null'], format: 'date-time' }, + }, + additionalProperties: false, + }, + }, + }, + additionalProperties: false, +}; + function humanize(path: string): string { return path .replace(/^\/v1\//, '') @@ -155,6 +183,7 @@ const PUBLIC_GET = [ '/spam/suspects', '/stats', '/stats/charts', + '/tags/recent', '/topology', '/v1', '/v1/exports/nodes.:format', @@ -168,6 +197,7 @@ const OWNER_GET = [ '/owner/csrf', '/owner/live', '/owner/live-last-hop', + '/owner/packet-sharing', '/owner/session', ] as const; @@ -181,10 +211,11 @@ export const API_CONTRACTS: readonly ApiContract[] = [ '/owner/alert-rules/:id/test', '/owner/login', '/owner/logout', + '/owner/packet-sharing', ], 'owner'), ...contracts('DELETE', ['/coverage/planned/:planId'], 'public'), ...contracts('DELETE', ['/owner/alert-rules/:id'], 'owner'), -].map((contract) => { +].map((contract): ApiContract => { if (contract.path === '/path-beta/slow-mode') { return { ...contract, @@ -250,6 +281,59 @@ export const API_CONTRACTS: readonly ApiContract[] = [ }, }; } + if (contract.path === '/owner/packet-sharing' && contract.method === 'GET') { + return { + ...contract, + summary: 'Read owner packet-sharing settings', + responseSchema: OWNER_PACKET_SHARING_SCHEMA, + queryParameters: [{ + name: 'nodeId', + in: 'query', + required: false, + schema: { type: 'string', pattern: '^[0-9A-Fa-f]{64}$' }, + }], + }; + } + if (contract.path === '/owner/packet-sharing' && contract.method === 'POST') { + return { + ...contract, + summary: 'Update owner packet-sharing settings', + requestSchema: { + type: 'object', + required: ['nodeId', 'enabled', 'destinationIds'], + properties: { + nodeId: { type: 'string', pattern: '^[0-9A-Fa-f]{64}$' }, + enabled: { type: 'boolean' }, + destinationIds: { + type: 'array', + maxItems: 32, + items: { type: 'string', pattern: '^[a-z0-9][a-z0-9_-]{0,63}$' }, + }, + }, + additionalProperties: false, + }, + responseSchema: { + type: 'object', + required: ['nodeId', 'enabled', 'destinationIds'], + properties: { + nodeId: { type: 'string', pattern: '^[0-9A-Fa-f]{64}$' }, + enabled: { type: 'boolean' }, + destinationIds: { + type: 'array', + maxItems: 32, + items: { type: 'string', pattern: '^[a-z0-9][a-z0-9_-]{0,63}$' }, + }, + }, + additionalProperties: false, + }, + additionalResponses: { + '409': { + description: 'One or more selected destinations are not configured', + responseSchema: ERROR_SCHEMA, + }, + }, + }; + } return contract; }); diff --git a/backend/src/api/routes.ts b/backend/src/api/routes.ts index f59be5b..716f831 100644 --- a/backend/src/api/routes.ts +++ b/backend/src/api/routes.ts @@ -25,11 +25,13 @@ import { getPacketDetail, getPublicVisibilityGeneration, getChannelMessageHistory, + getRecentMessageTags, getRecentPacketEvents, getRecentPackets, analyticsQuery, pool, query, + withTransaction, } from '../db/index.js'; import { autoLinkOwnerNodeIds, @@ -188,6 +190,7 @@ registerMiscRoutes(router, { getRecentPacketEvents, getPacketDetail, getChannelMessageHistory, + getRecentMessageTags, getPublicVisibilityGeneration, packetDetailLimiter: PACKET_DETAIL_LIMITER, }); @@ -213,6 +216,7 @@ registerOwnerRoutes(router, { getOwnerCredentialGeneration, invalidateOwnerNodeIdCache, query, + withTransaction, }); registerPathingRoutes(router, { pathBetaLimiter: PATH_BETA_LIMITER, diff --git a/backend/src/api/routes/misc.ts b/backend/src/api/routes/misc.ts index 713fb5e..4b74a01 100644 --- a/backend/src/api/routes/misc.ts +++ b/backend/src/api/routes/misc.ts @@ -51,6 +51,7 @@ type GetChannelMessageHistoryFn = ( network?: string, observer?: string, ) => Promise; +type GetRecentMessageTagsFn = (limit: number, network?: string) => Promise; type MiscRouteDeps = { query: QueryFn; @@ -58,6 +59,7 @@ type MiscRouteDeps = { getRecentPacketEvents: GetRecentPacketEventsFn; getPacketDetail: GetPacketDetailFn; getChannelMessageHistory: GetChannelMessageHistoryFn; + getRecentMessageTags: GetRecentMessageTagsFn; getPublicVisibilityGeneration: () => Promise; packetDetailLimiter: ReturnType; }; @@ -69,6 +71,7 @@ export function registerMiscRoutes(router: Router, deps: MiscRouteDeps): void { getRecentPacketEvents, getPacketDetail, getChannelMessageHistory, + getRecentMessageTags, getPublicVisibilityGeneration, packetDetailLimiter, } = deps; @@ -137,6 +140,27 @@ export function registerMiscRoutes(router: Router, deps: MiscRouteDeps): void { } }); + // Message tags (kind / speaker / topic / flags) for live feed enrichment. + // Tags are written ~2s after a packet by tagger-worker; the frontend patches + // rows by hash. Public visibility + network scope enforced in the query. + router.get('/tags/recent', async (req, res) => { + const limit = parseBoundedInteger(req.query['limit'], { + name: 'limit', + defaultValue: 200, + min: 1, + max: 500, + }); + try { + const network = resolvePublicNetworkScope(req.query['network'], req.headers); + const tags = await getRecentMessageTags(limit, network); + res.setHeader('Cache-Control', 'no-store'); + res.json(tags); + } catch (err) { + console.error('[api] GET /tags/recent', (err as Error).message); + res.status(500).json({ error: 'Internal server error' }); + } + }); + router.get('/packets/:hash', packetDetailLimiter, async (req, res) => { const hash = parseHexIdentifier(req.params['hash'], { name: 'hash', diff --git a/backend/src/db/index.ts b/backend/src/db/index.ts index d8d1f16..1541edc 100644 --- a/backend/src/db/index.ts +++ b/backend/src/db/index.ts @@ -149,6 +149,11 @@ async function queryPool( } } +export type DatabaseQueryFn = ( + text: string, + params?: unknown[], +) => Promise<{ rows: T[] }>; + export async function query( text: string, params?: unknown[], @@ -157,6 +162,28 @@ export async function query( return queryPool(pool, 'oltp', text, params, signal); } +export async function withTransaction(work: (transactionQuery: DatabaseQueryFn) => Promise): Promise { + const client = await pool.connect(); + const transactionQuery: DatabaseQueryFn = async ( + text: string, + params?: unknown[], + ) => { + const result = await client.query(text, params); + return { rows: result.rows }; + }; + try { + await client.query('BEGIN'); + const result = await work(transactionQuery); + await client.query('COMMIT'); + return result; + } catch (error) { + await client.query('ROLLBACK').catch(() => undefined); + throw error; + } finally { + client.release(); + } +} + export async function namedQuery( name: string, text: string, @@ -771,17 +798,35 @@ export async function getNodes( export async function getNodeHistory(nodeId: string, hours = 24, network = 'ukmesh') { const scope = buildScopePlaceholders(3, network); + const aliases = nodeAliasArraySql('$1'); + // Both consumers describe general activity, including for observers that + // are also repeaters. Include receptions and known origins for every role. + // Drive each direction with the canonical alias array and an equality + // lookup: a combined OR/ANY lets LIMIT choose a scan of time-ordered traffic + // for quiet nodes. Per-alias limits keep both node/time indexes useful even + // at 28 days; the final limit retains the newest 500 observations overall. + const packetsFor = (column: 'rx_node_id' | 'src_node_id') => ` + SELECT p.time, p.packet_hash, + COALESCE(src_alias.canonical_node_id, UPPER(BTRIM(p.src_node_id))) AS src_node_id, + topic, packet_type, hop_count, rssi, snr, payload + FROM packets p + LEFT JOIN node_identity_aliases src_alias + ON src_alias.source_node_id = UPPER(BTRIM(p.src_node_id)) + WHERE p.${column} = requested.node_id + ${column === 'src_node_id' ? `AND NOT COALESCE(p.rx_node_id = ANY(${aliases}), FALSE)` : ''} + AND p.time > NOW() - INTERVAL '1 hour' * $2 + AND p.time <= NOW() + ${buildPacketScopeClause(scope, 'p', network)} + ${buildPublicPacketPrivacyClause('p')} + ORDER BY time DESC LIMIT 500`; const res = await pool.query( - `SELECT p.time, p.packet_hash, - COALESCE(src_alias.canonical_node_id, UPPER(BTRIM(p.src_node_id))) AS src_node_id, - topic, packet_type, hop_count, rssi, snr, payload - FROM packets p - LEFT JOIN node_identity_aliases src_alias - ON src_alias.source_node_id = UPPER(BTRIM(p.src_node_id)) - WHERE p.rx_node_id = ANY(${nodeAliasArraySql('$1')}) - AND p.time > NOW() - INTERVAL '1 hour' * $2 - ${buildPacketScopeClause(scope, 'p', network)} - ${buildPublicPacketPrivacyClause('p')} + `SELECT recent.* + FROM unnest(${aliases}) requested(node_id) + CROSS JOIN LATERAL ( + (${packetsFor('rx_node_id')}) + UNION ALL + (${packetsFor('src_node_id')}) + ) recent ORDER BY time DESC LIMIT 500`, [nodeId, hours, ...scope.params] ); @@ -902,9 +947,11 @@ export async function getRecentPackets( ${fields === 'full' ? 'rp.topic, rp.topic_prefix, rp.iata, rp.route_type, rp.network, rp.transport_codes, rp.region_scope,' : ''} - ps.observer_node_ids, ps.observer_iatas, ps.rx_count, ps.tx_count + ps.observer_node_ids, ps.observer_iatas, ps.rx_count, ps.tx_count, + mt.tags, mt.confidence AS tag_confidence FROM recent_packets rp LEFT JOIN packet_stats ps ON ps.packet_hash = rp.packet_hash + LEFT JOIN message_tags mt ON mt.packet_hash = rp.packet_hash ORDER BY rp.time DESC LIMIT $1`, [limit, ...scope.params], @@ -969,9 +1016,11 @@ export async function getRecentMessages( m.time, m.packet_hash, m.rx_node_id, m.src_node_id, m.topic, m.iata, m.packet_type, m.hop_count, m.rssi, m.snr, m.payload, m.summary, m.advert_count, m.path_hashes, m.path_hash_size_bytes, - ms.observer_node_ids, ms.observer_iatas, ms.rx_count, ms.tx_count + ms.observer_node_ids, ms.observer_iatas, ms.rx_count, ms.tx_count, + mt.tags, mt.confidence AS tag_confidence FROM recent_msgs m LEFT JOIN msg_stats ms ON ms.packet_hash = m.packet_hash + LEFT JOIN message_tags mt ON mt.packet_hash = m.packet_hash ORDER BY m.time DESC LIMIT $1`, [limit, ...scope.params], @@ -980,6 +1029,32 @@ export async function getRecentMessages( return res.rows; } +/** + * Recent message tags for live feed enrichment. Tags are produced by the + * tagger-worker a few seconds after a packet arrives, so the live WS feed can + * never carry them inline; the frontend polls this and patches rows by hash. + * Only tags whose packet is publicly visible and in scope are returned. + */ +export async function getRecentMessageTags(limit = 200, network?: string) { + const scope = buildScopePlaceholders(2, network); + const res = await query( + `SELECT mt.packet_hash, mt.tags, mt.confidence, mt.tagged_at + FROM message_tags mt + WHERE mt.tagged_at > NOW() - INTERVAL '24 hours' + AND EXISTS ( + SELECT 1 FROM packets p + WHERE p.packet_hash = mt.packet_hash + AND p.time > NOW() - INTERVAL '24 hours' + ${buildPacketScopeClause(scope, 'p', network)} + ${buildPublicPacketPrivacyClause('p')} + ) + ORDER BY mt.tagged_at DESC + LIMIT $1`, + [limit, ...scope.params], + ); + return res.rows; +} + /** * Fetch one channel's decrypted GroupText history. The WebSocket snapshot stays * intentionally recent and bounded; this demand-driven read reaches back 90 @@ -1089,8 +1164,10 @@ export async function getChannelMessageHistory( CASE WHEN m.rx_node_id IS NOT NULL THEN ARRAY[m.rx_node_id]::text[] ELSE ARRAY[]::text[] END AS observer_node_ids, CASE WHEN NULLIF(TRIM(m.iata), '') IS NOT NULL THEN ARRAY[m.iata]::text[] ELSE ARRAY[]::text[] END AS observer_iatas, CASE WHEN COALESCE(m.payload->>'direction', 'rx') <> 'tx' THEN 1 ELSE 0 END::int AS rx_count, - CASE WHEN COALESCE(m.payload->>'direction', 'rx') = 'tx' THEN 1 ELSE 0 END::int AS tx_count + CASE WHEN COALESCE(m.payload->>'direction', 'rx') = 'tx' THEN 1 ELSE 0 END::int AS tx_count, + mt.tags, mt.confidence AS tag_confidence FROM channel_candidates m + LEFT JOIN message_tags mt ON mt.packet_hash = m.packet_hash WHERE 1 = 1 ${buildPublicPacketPrivacyClause('m')} ORDER BY m.time DESC`, @@ -1107,9 +1184,14 @@ export async function getChannelMessageHistory( }); } -export async function getRecentPacketEvents(limit = 200, network?: string, observer?: string) { +export async function getRecentPacketEvents(limit = 200, network?: string, observer?: string, regionScope?: string) { const scope = buildScopePlaceholders(2, network, observer); const params: unknown[] = [limit, ...scope.params]; + let regionClause = ''; + if (regionScope) { + params.push(regionScope); + regionClause = `AND p.region_scope = $${params.length}`; + } const res = await pool.query( `SELECT p.time, p.packet_hash, @@ -1142,8 +1224,10 @@ export async function getPacketDetail(hash: string, network = 'ukmesh') { pool.query( `SELECT p.time, p.packet_hash, p.rx_node_id, p.src_node_id, p.topic, p.iata, p.packet_type, p.route_type, p.hop_count, p.rssi, p.snr, - p.payload, p.path_hashes, p.path_hash_size_bytes, p.raw_hex + p.payload, p.path_hashes, p.path_hash_size_bytes, p.raw_hex, + mt.tags, mt.confidence AS tag_confidence FROM packets p + LEFT JOIN message_tags mt ON mt.packet_hash = p.packet_hash WHERE p.packet_hash = $1 ${buildPacketScopeClause(scope, 'p', network)} ${buildPublicPacketPrivacyClause('p')} @@ -1185,6 +1269,8 @@ export async function getPacketDetail(hash: string, network = 'ukmesh') { pathHashes: row.path_hashes as string[] | null, pathHashSizeBytes: row.path_hash_size_bytes as number | null, rawHex: row.raw_hex as string | null, + tags: (row.tags as Record | null) ?? null, + tagConfidence: (row.tag_confidence as Record | null) ?? null, observations: observations.rows.map((r) => ({ rxNodeId: r.rx_node_id as string | null, iata: r.iata as string | null, diff --git a/backend/src/ws/server.ts b/backend/src/ws/server.ts index 944e527..032eb54 100644 --- a/backend/src/ws/server.ts +++ b/backend/src/ws/server.ts @@ -110,6 +110,8 @@ function slimInitialPacket(raw: InitialStateRow): InitialStateRow { path_hash_size_bytes: raw['path_hash_size_bytes'], rx_count: raw['rx_count'], tx_count: raw['tx_count'], + tags: raw['tags'], + tag_confidence: raw['tag_confidence'], }; } diff --git a/docs/openapi.yaml b/docs/openapi.yaml index 21aa73c..c02b2ea 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -619,6 +619,64 @@ } } }, + "/api/tags/recent": { + "get": { + "summary": "Read recent message tags for live feed enrichment", + "operationId": "get-tags-recent", + "tags": [ + "public" + ], + "x-access": "public", + "parameters": [ + { + "name": "limit", + "in": "query", + "required": false, + "schema": { + "type": "integer", + "minimum": 1, + "maximum": 500, + "default": 200 + } + } + ], + "responses": { + "200": { + "description": "Successful response", + "content": { + "application/json": { + "schema": { + "type": "array", + "maxItems": 500, + "items": { + "type": "object", + "required": [ + "packet_hash", + "tags" + ], + "properties": { + "packet_hash": { + "type": "string" + }, + "tags": { + "type": "object" + }, + "confidence": { + "type": "object" + }, + "tagged_at": { + "type": "string", + "format": "date-time" + } + } + } + } + } + } + } + } + } + }, "/api/feed/messages": { "get": { "summary": "Read historical messages for one channel", @@ -1809,6 +1867,87 @@ } } }, + "/api/nodes/top-adverts": { + "get": { + "summary": "Read nodes top adverts", + "operationId": "get-nodes-top-adverts", + "tags": [ + "public" + ], + "x-access": "public", + "parameters": [], + "responses": { + "200": { + "description": "Successful response", + "content": { + "application/json": { + "schema": { + "type": "object", + "additionalProperties": true + } + } + } + }, + "400": { + "description": "Invalid or out-of-bounds request", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "429": { + "description": "Rate limit exceeded", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "500": { + "description": "Internal server error", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + } + } + } + }, "/api/observer-activity": { "get": { "summary": "Read observer activity", @@ -3338,6 +3477,409 @@ ] } }, + "/api/owner/packet-sharing": { + "get": { + "summary": "Read owner packet-sharing settings", + "operationId": "get-owner-packet-sharing", + "tags": [ + "owner" + ], + "x-access": "owner", + "parameters": [ + { + "name": "nodeId", + "in": "query", + "required": false, + "schema": { + "type": "string", + "pattern": "^[0-9A-Fa-f]{64}$" + } + } + ], + "responses": { + "200": { + "description": "Successful response", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "nodeId", + "featureEnabled", + "enabled", + "destinations" + ], + "properties": { + "nodeId": { + "type": "string", + "pattern": "^[0-9A-Fa-f]{64}$" + }, + "featureEnabled": { + "type": "boolean" + }, + "enabled": { + "type": "boolean" + }, + "destinations": { + "type": "array", + "items": { + "type": "object", + "required": [ + "id", + "name", + "websiteUrl", + "description", + "configured", + "selected", + "lastForwardedAt" + ], + "properties": { + "id": { + "type": "string", + "pattern": "^[a-z0-9][a-z0-9_-]{0,63}$" + }, + "name": { + "type": "string", + "minLength": 1, + "maxLength": 120 + }, + "websiteUrl": { + "type": [ + "string", + "null" + ], + "format": "uri", + "maxLength": 2048 + }, + "description": { + "type": [ + "string", + "null" + ], + "maxLength": 160 + }, + "configured": { + "type": "boolean" + }, + "selected": { + "type": "boolean" + }, + "lastForwardedAt": { + "type": [ + "string", + "null" + ], + "format": "date-time" + } + }, + "additionalProperties": false + } + } + }, + "additionalProperties": false + } + } + } + }, + "400": { + "description": "Invalid or out-of-bounds request", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "401": { + "description": "Authentication required", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "403": { + "description": "Authentication or CSRF validation failed", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "429": { + "description": "Rate limit exceeded", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "500": { + "description": "Internal server error", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + } + }, + "security": [ + { + "ownerSession": [] + } + ] + }, + "post": { + "summary": "Update owner packet-sharing settings", + "operationId": "post-owner-packet-sharing", + "tags": [ + "owner" + ], + "x-access": "owner", + "parameters": [], + "responses": { + "200": { + "description": "Successful response", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "nodeId", + "enabled", + "destinationIds" + ], + "properties": { + "nodeId": { + "type": "string", + "pattern": "^[0-9A-Fa-f]{64}$" + }, + "enabled": { + "type": "boolean" + }, + "destinationIds": { + "type": "array", + "maxItems": 32, + "items": { + "type": "string", + "pattern": "^[a-z0-9][a-z0-9_-]{0,63}$" + } + } + }, + "additionalProperties": false + } + } + } + }, + "400": { + "description": "Invalid or out-of-bounds request", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "401": { + "description": "Authentication required", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "403": { + "description": "Authentication or CSRF validation failed", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "409": { + "description": "One or more selected destinations are not configured", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "429": { + "description": "Rate limit exceeded", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "500": { + "description": "Internal server error", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + } + }, + "security": [ + { + "ownerSession": [] + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "nodeId", + "enabled", + "destinationIds" + ], + "properties": { + "nodeId": { + "type": "string", + "pattern": "^[0-9A-Fa-f]{64}$" + }, + "enabled": { + "type": "boolean" + }, + "destinationIds": { + "type": "array", + "maxItems": 32, + "items": { + "type": "string", + "pattern": "^[a-z0-9][a-z0-9_-]{0,63}$" + } + } + }, + "additionalProperties": false + } + } + } + } + } + }, "/api/owner/session": { "get": { "summary": "Read owner session", diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 2308e8b..39312be 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -16,6 +16,7 @@ import { rfNodeCoverageState, useRfCoverage, type RfCoverageTierName } from './h import { useDashboardStats, type DashboardStats } from './hooks/useDashboardStats.js'; import { linkStateStore } from './hooks/useLinkState.js'; import { useAppMessageHandler } from './hooks/useAppMessageHandler.js'; +import { useMessageTags } from './hooks/useMessageTags.js'; import { HEATMAP_CAPABLE, PACKET_ARCS_CAPABLE, @@ -469,6 +470,8 @@ export const App: React.FC = () => { { network: networkFilter, observer: observerFilter }, scopeState.nodeEpoch, ); + // Live tag enrichment for the feed panel (tags land ~2s after a packet). + useMessageTags({ network: networkFilter, observer: observerFilter }, realtimeScopeKey); const wsState = wsConnection.readyState; wsConnectedRef.current = wsState === 'connected'; diff --git a/frontend/src/components/PacketFeed.tsx b/frontend/src/components/PacketFeed.tsx index e8d1ce6..288be05 100644 --- a/frontend/src/components/PacketFeed.tsx +++ b/frontend/src/components/PacketFeed.tsx @@ -1,6 +1,6 @@ import React, { useEffect, useRef, useState } from 'react'; import { useMessages, useNodeMap } from '../hooks/useNodes.js'; -import { aggregatedPacketObserverIataLabel } from '../hooks/packetFeed.js'; +import { aggregatedPacketObserverIataLabel, messageTagKind, messageTagTitle } from '../hooks/packetFeed.js'; import { useOverlayStore } from '../store/overlayStore.js'; import { useWatchlist } from '../hooks/useWatchlist.js'; import type { AggregatedPacket } from '../hooks/useNodes.js'; @@ -91,6 +91,11 @@ const PacketFeedItem: React.FC = React.memo(({ } + {messageTagKind(p) && ( + + {messageTagKind(p)} + + )}
+ {/* Message tags (tagger-worker / Jev) */} + {(() => { + const tags = (packet.tags ?? detail?.tags) as MessageTags | undefined | null; + if (!tags || Object.keys(tags).length === 0) return null; + const conf = (packet.tag_confidence ?? detail?.tagConfidence) as TagConfidence | undefined | null; + const chips: string[] = []; + if (tags.kind) chips.push(tags.kind); + if (tags.topic) chips.push(`topic: ${tags.topic}`); + if (tags.speaker) chips.push(`speaker: ${tags.speaker}`); + if ((tags.safety ?? 0) >= 0.5) chips.push('⚠ safety'); + if ((tags.directed ?? 0) >= 0.5) chips.push('directed'); + if ((tags.mentions_location ?? 0) >= 0.5) chips.push('mentions location'); + return ( +
+
+ Message tags{typeof conf?.kind === 'number' ? ` · kind ${(conf.kind * 100).toFixed(0)}%` : ''} +
+
+ {chips.map((chip) => ( + {chip} + ))} +
+
+ ); + })()} + {/* Observer table */} {detail?.observations && detail.observations.length > 0 && (
diff --git a/frontend/src/pages/ukmesh/UKFeedPage.tsx b/frontend/src/pages/ukmesh/UKFeedPage.tsx index 9053681..1fa0806 100644 --- a/frontend/src/pages/ukmesh/UKFeedPage.tsx +++ b/frontend/src/pages/ukmesh/UKFeedPage.tsx @@ -9,6 +9,8 @@ import { } from '../../hooks/useNodes.js'; import { mapMessageRows, type RecentPacketRow } from '../../hooks/packetFeed.js'; import { useAppMessageHandler } from '../../hooks/useAppMessageHandler.js'; +import { useMessageTags } from '../../hooks/useMessageTags.js'; +import { messageTagKind, messageTagTitle } from '../../hooks/packetFeed.js'; import { ApiResponseError, chartStatsEndpoint, @@ -114,6 +116,9 @@ export const UKFeedPage: React.FC = () => { const wsConnection = useWebSocket(handleWSMessage, scope, scopeEpoch); + // Live tag enrichment: tags land ~2s after a packet, so patch rows by hash. + useMessageTags(scope, scopeKey); + useEffect(() => { historyRequestRef.current?.abort(); historyRequestRef.current = null; @@ -315,6 +320,9 @@ export const UKFeedPage: React.FC = () => { return filteredPackets.slice(0, MAX_PACKETS); }, [filteredPackets]); const virtualRows = useMemo(() => { + // Rows are natural-height; tagged rows add one ~18px chips line under the + // meta section. Keep the estimate aligned with the taller (tagged) case so + // spacers never underestimate and rows cannot overlap (2026-09-17). const rowHeight = 76; const visibleHeight = Math.max(320, viewportHeight - 200); const start = Math.max(0, Math.floor(listScrollTop / rowHeight) - 5); @@ -653,6 +661,13 @@ export const UKFeedPage: React.FC = () => { {packet.packet_hash} {observerDisplay}
+ {messageTagKind(packet) && ( +
+ + {messageTagKind(packet)} + +
+ )}

{packetSummary(packet, nodeMap)}

@@ -689,6 +704,11 @@ export const UKFeedPage: React.FC = () => { {selectedPacket.hop_count} hop{selectedPacket.hop_count !== 1 ? 's' : ''} )} + {messageTagKind(selectedPacket) && ( + + {messageTagKind(selectedPacket)} + + )}

{packetSummary(selectedPacket, nodeMap)}

@@ -748,6 +768,11 @@ export const UKFeedPage: React.FC = () => { {selectedPacket.hop_count} hop{selectedPacket.hop_count !== 1 ? 's' : ''} )} + {messageTagKind(selectedPacket) && ( + + {messageTagKind(selectedPacket)} + + )}

{packetSummary(selectedPacket, nodeMap)}

diff --git a/frontend/src/pages/ukmesh/feed-page.css b/frontend/src/pages/ukmesh/feed-page.css index 293dbfc..0b2d21e 100644 --- a/frontend/src/pages/ukmesh/feed-page.css +++ b/frontend/src/pages/ukmesh/feed-page.css @@ -1173,3 +1173,30 @@ border-bottom: 1px solid var(--border); } } + +/* ── Message tags (tagger-worker / Jev) ────────────────────────────────────── */ +.uk-feed-packet-row__tags { + display: flex; + flex-wrap: wrap; + gap: 4px; + margin: 2px 0 0; +} +.msg-tag-chip { + display: inline-flex; + align-items: center; + padding: 0 6px; + border-radius: 999px; + border: 1px solid rgba(38, 198, 162, 0.45); + background: rgba(38, 198, 162, 0.12); + color: #9fe8d6; + font-size: 10px; + line-height: 16px; + letter-spacing: 0.04em; + text-transform: uppercase; + white-space: nowrap; +} +.feed-detail__tags { + display: flex; + flex-wrap: wrap; + gap: 6px; +} diff --git a/frontend/src/pages/ukmesh/feedModel.ts b/frontend/src/pages/ukmesh/feedModel.ts index edaeaf0..650d9f7 100644 --- a/frontend/src/pages/ukmesh/feedModel.ts +++ b/frontend/src/pages/ukmesh/feedModel.ts @@ -1,4 +1,5 @@ import type { AggregatedPacket, MeshNode } from '../../hooks/useNodes.js'; +import type { MessageTags, TagConfidence } from '../../hooks/packetFeed.js'; import { ScopedCache } from '../../utils/scopedCache.js'; import type { LazyPathNode, LazyPathResult } from './PacketDetailPanel.js'; @@ -21,6 +22,8 @@ export type FeedPacket = { tx_count?: number; summary?: string | null; path_hashes?: string[] | null; + tags?: MessageTags; + tag_confidence?: TagConfidence; }; export const TYPE_LABELS: Record = { @@ -217,6 +220,8 @@ export function aggregatedPacketToFeedPacket(packet: AggregatedPacket): FeedPack tx_count: packet.txCount, summary: packet.summary ?? null, path_hashes: (packet.path as string[] | undefined) ?? null, + tags: packet.tags, + tag_confidence: packet.tagConfidence, }; } @@ -251,7 +256,7 @@ export function mergeFeedPacketObservations(current: FeedPacket, next: FeedPacke ...(next.observer_iatas ?? []), next.iata, ]); - return { + const merged: FeedPacket = { ...latest, first_seen_time: firstSeenCandidates.length > 0 ? new Date(Math.min(...firstSeenCandidates)).toISOString() @@ -267,6 +272,13 @@ export function mergeFeedPacketObservations(current: FeedPacket, next: FeedPacke rx_count: Math.max(current.rx_count ?? 0, next.rx_count ?? 0), tx_count: Math.max(current.tx_count ?? 0, next.tx_count ?? 0), }; + // Tags only exist once the tagger has run; keep the key absent when untagged + // so feed-row shapes stay stable for callers and tests. + const tags = latest.tags ?? current.tags ?? next.tags; + if (tags !== undefined) merged.tags = tags; + const tagConfidence = latest.tag_confidence ?? current.tag_confidence ?? next.tag_confidence; + if (tagConfidence !== undefined) merged.tag_confidence = tagConfidence; + return merged; } export function packetObserverIatas(packet: FeedPacket, nodeMap: Map): string[] { diff --git a/frontend/src/styles/map-app.css b/frontend/src/styles/map-app.css index 2a092a2..03320d2 100644 --- a/frontend/src/styles/map-app.css +++ b/frontend/src/styles/map-app.css @@ -2220,3 +2220,28 @@ top: calc(48px + 56px + 8px); } } + +/* ── Message tags (tagger-worker / Jev) ────────────────────────────────────── */ +.msg-tag-chip { + display: inline-flex; + align-items: center; + padding: 0 6px; + border-radius: 999px; + border: 1px solid rgba(38, 198, 162, 0.45); + background: rgba(38, 198, 162, 0.12); + color: #9fe8d6; + font-size: 10px; + line-height: 16px; + letter-spacing: 0.04em; + text-transform: uppercase; + white-space: nowrap; +} +.packet-item__tag { + margin-left: 2px; +} +.packet-item__tags { + display: flex; + flex-wrap: wrap; + gap: 4px; + margin-top: 2px; +} diff --git a/tagger-worker/tagger_worker.py b/tagger-worker/tagger_worker.py new file mode 100644 index 0000000..ee5c548 --- /dev/null +++ b/tagger-worker/tagger_worker.py @@ -0,0 +1,347 @@ +#!/usr/bin/env python3 +"""UKMesh realtime message tagger - TypeSafe/Jev. + +Polls the packets table for new public-channel messages, asks Jev to tag each +one (speaker / kind / topic + flags), and stores the result in message_tags. +Tags are displayed on the public feed and app.ukmesh.com detail panels; private +or out-of-scope packets are never sent to the API (public visibility predicate + +network scope applied in the query). Monitoring data lives in tagger_state +(heartbeat + counters). + +Failure handling: a message that fails all inline API attempts is persisted to +tagger_retry and retried on every subsequent poll (up to TAGGER_MAX_RETRY_ATTEMPTS) +so a TypeSafe outage cannot silently drop tags (2026-09-17 hardening; the old +20s overlap window lost 6 messages during two 503 bursts). +""" +import json +import os +import signal +import sys +import time +import urllib.error +import urllib.request + +import psycopg2 + +API_URL = os.environ.get('TYPESAFE_API_URL', 'https://api.typesafe.ai/v1/systemone') +API_KEY = os.environ.get('TYPESAFE_API_KEY', '').strip() +DATABASE_URL = os.environ['DATABASE_URL'] +POLL_S = float(os.environ.get('TAGGER_POLL_S', '5')) +CALL_FLOOR_S = float(os.environ.get('TAGGER_CALL_FLOOR_S', '0.5')) +BATCH_LIMIT = int(os.environ.get('TAGGER_BATCH_LIMIT', '200')) +OVERLAP_S = int(os.environ.get('TAGGER_OVERLAP_S', '120')) +MAX_RETRY_ATTEMPTS = int(os.environ.get('TAGGER_MAX_RETRY_ATTEMPTS', '30')) +RETRY_BATCH = int(os.environ.get('TAGGER_RETRY_BATCH', '20')) +MODEL = os.environ.get('TYPESAFE_MODEL', 'jev-latest') +START = os.environ.get('TAGGER_START', '').strip() +NETWORKS = [n.strip() for n in os.environ.get('TAGGER_NETWORKS', 'ukmesh,northeast,teesside').split(',') if n.strip()] + +QUESTIONS = { + "speaker": {"type": "choice", "instructions": "Who most likely sent this message?", + "criteria": {"human": "A human operator typing a message", + "bot": "An automated bot or script", + "room-server": "A room server, repeater, or bridge auto-response", + "unclear": "Cannot tell from the message"}}, + "kind": {"type": "choice", "instructions": "What is the primary kind of this message?", + "criteria": {"chat": "Social conversation, greeting, banter, or sharing (links, pictures, files, images)", + "radio-check": "Checking whether/how their signal is received", + "test-ping": "A test or ping", + "ack": "Acknowledgement of a previous signal", + "position": "Sharing their whereabouts or movements (e.g. 'I am near X', 'heading to Y', 'going off air for a bit') - not announcements about infrastructure or network status", + "coverage": "Reporting range, hops, or propagation", + "help": "Asking for help or information from others", + "noise": "No meaningful content (gibberish, stray characters, emoji only)", + "spam": "Only unsolicited promotion, scams, or abuse. IMPORTANT: a bare link, picture, or file with no promotional text is NEVER spam (mesh users share links/images routinely); when choosing between spam and chat for a bare link, choose chat", + "automation": "Automated telemetry, home-automation, or sensor notification", + "command": "A command to a system or bot (e.g. !test, !path, !help)"}}, + "topic": {"type": "choice", "instructions": "What is this message mainly about?", + "criteria": {"weather": "Weather or conditions", + "meetups": "Meetups, events, or social plans", + "travel": "Travel, journeys, or being on the move", + "gear": "Radios, antennas, equipment, or tech gear", + "network": "Network coverage, repeaters, hops, or connectivity", + "help": "Getting help, support, or answering questions", + "humour": "Jokes, banter, or humour", + "general": "General chatter that fits nothing else"}}, + "mentions_location": {"type": "noul", "instructions": "Does the message mention a geographic place (town, area, or landmark)?"}, + "directed": {"type": "noul", "instructions": "Is the message addressed to a specific person or node (by name or @mention)?"}, + "safety": {"type": "noul", "instructions": "Is this message safety or emergency related?"}, +} + +DDL = """ +CREATE TABLE IF NOT EXISTS message_tags ( + packet_hash text PRIMARY KEY, + packet_time timestamptz NOT NULL, + tagged_at timestamptz NOT NULL DEFAULT now(), + model text, + latency_ms integer, + tags jsonb NOT NULL, + confidence jsonb +); +CREATE TABLE IF NOT EXISTS tagger_state ( + id int PRIMARY KEY CHECK (id = 1), + cursor_time timestamptz, + seen bigint NOT NULL DEFAULT 0, + tagged bigint NOT NULL DEFAULT 0, + errors bigint NOT NULL DEFAULT 0, + http_429 bigint NOT NULL DEFAULT 0, + last_error text, + avg_latency_ms_50 real, + started_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz +); +INSERT INTO tagger_state (id) VALUES (1) ON CONFLICT DO NOTHING; +ALTER TABLE tagger_state ADD COLUMN IF NOT EXISTS retried bigint NOT NULL DEFAULT 0; +ALTER TABLE tagger_state ADD COLUMN IF NOT EXISTS given_up bigint NOT NULL DEFAULT 0; +CREATE TABLE IF NOT EXISTS tagger_retry ( + packet_hash text PRIMARY KEY, + packet_time timestamptz NOT NULL, + attempts int NOT NULL DEFAULT 0, + last_error text, + first_failed_at timestamptz NOT NULL DEFAULT now(), + exhausted boolean NOT NULL DEFAULT false +); +""" + +_running = True + + +def _stop(*_a): + global _running + _running = False + + +signal.signal(signal.SIGTERM, _stop) +signal.signal(signal.SIGINT, _stop) + + +def log(msg): + print(f"[tagger] {msg}", flush=True) + + +def api_tag(state): + payload = {'state': state, 'model': MODEL, 'questions': QUESTIONS} + req = urllib.request.Request(API_URL, data=json.dumps(payload).encode(), + headers={'Authorization': f'Bearer {API_KEY}', + 'Content-Type': 'application/json'}) + t0 = time.time() + with urllib.request.urlopen(req, timeout=90) as r: + resp = json.loads(r.read().decode()) + return resp, (time.time() - t0) * 1000.0 + + +def flatten(answers): + tags, conf = {}, {} + for k, v in (answers or {}).items(): + if not isinstance(v, dict): + continue + if 'choice' in v: + tags[k] = v.get('choice') + elif 'noul' in v: + tags[k] = v.get('noul') + elif 'score' in v: + tags[k] = v.get('score') + if 'confidence' in v: + conf[k] = v.get('confidence') + return tags, conf + + +def tag_message(cur, ph, ts): + """Tag one message with up to 3 inline attempts. Returns (ok, latency_ms, err).""" + cur.execute(""" + SELECT payload->'decrypted'->>'sender', payload->'decrypted'->>'message' + FROM packets WHERE packet_hash=%s LIMIT 1 + """, (ph,)) + row = cur.fetchone() + if row is None or row[1] is None: + return True, None, None # nothing taggable (row vanished / no message) + sender, message = row + state = f"Message from node '{sender or 'unknown'}': \"{(message or '')[:4000]}\"" + for attempt in (1, 2, 3): + try: + resp, ms = api_tag(state) + tags, conf = flatten(resp.get('answers')) + cur.execute("""INSERT INTO message_tags + (packet_hash, packet_time, tagged_at, model, latency_ms, tags, confidence) + VALUES (%s, %s, now(), %s, %s, %s, %s) + ON CONFLICT (packet_hash) DO NOTHING""", + (ph, ts, MODEL, int(ms), json.dumps(tags), json.dumps(conf))) + return True, ms, None + except urllib.error.HTTPError as e: + if e.code == 429: + cur.execute("UPDATE tagger_state SET http_429=http_429+1 WHERE id=1") + time.sleep(min(CALL_FLOOR_S * (2 ** attempt), 30)) + continue + err = f'http {e.code}: {e.read().decode()[:160]}' + time.sleep(2 * attempt) + except Exception as ex: # noqa: BLE001 + err = str(ex)[:200] + time.sleep(2 * attempt) + return False, None, err + + +def process_retries(cur): + """Retry previously failed messages. Returns (ok_count, fail_count, pending).""" + cur.execute(""" + SELECT packet_hash, packet_time, attempts FROM tagger_retry + WHERE NOT exhausted AND attempts < %s + ORDER BY first_failed_at ASC LIMIT %s + """, (MAX_RETRY_ATTEMPTS, RETRY_BATCH)) + due = cur.fetchall() + ok = fail = 0 + for ph, ts, attempts in due: + cur.execute("SELECT 1 FROM message_tags WHERE packet_hash=%s", (ph,)) + if cur.fetchone(): + cur.execute("DELETE FROM tagger_retry WHERE packet_hash=%s", (ph,)) + continue + good, ms, err = tag_message(cur, ph, ts) + if good: + cur.execute("DELETE FROM tagger_retry WHERE packet_hash=%s", (ph,)) + cur.execute("UPDATE tagger_state SET retried=retried+1 WHERE id=1") + ok += 1 + else: + fail += 1 + if attempts + 1 >= MAX_RETRY_ATTEMPTS: + cur.execute("""UPDATE tagger_retry SET attempts=attempts+1, last_error=%s, + exhausted=true WHERE packet_hash=%s""", (err, ph)) + cur.execute("UPDATE tagger_state SET given_up=given_up+1 WHERE id=1") + else: + cur.execute("""UPDATE tagger_retry SET attempts=attempts+1, last_error=%s + WHERE packet_hash=%s""", (err, ph)) + time.sleep(CALL_FLOOR_S) + cur.execute("SELECT count(*) FROM tagger_retry WHERE NOT exhausted") + pending = cur.fetchone()[0] + return ok, fail, pending + + +def main(): + if not API_KEY: + log('FATAL: TYPESAFE_API_KEY is empty') + sys.exit(2) + conn = psycopg2.connect(DATABASE_URL) + conn.autocommit = True + cur = conn.cursor() + cur.execute(DDL) + cur.execute("SELECT cursor_time FROM tagger_state WHERE id=1") + cursor = cur.fetchone()[0] + if cursor is None: + if START: + cur.execute("UPDATE tagger_state SET cursor_time=%s::timestamptz WHERE id=1", (START,)) + else: + cur.execute("UPDATE tagger_state SET cursor_time=now() - interval '5 minutes' WHERE id=1") + cur.execute("SELECT cursor_time FROM tagger_state WHERE id=1") + cursor = cur.fetchone()[0] + log(f"started; model={MODEL} poll={POLL_S}s floor={CALL_FLOOR_S}s overlap={OVERLAP_S}s " + f"networks={','.join(NETWORKS)} cursor={cursor.isoformat()}") + + lat_hist = [] + last_log = 0.0 + while _running: + loop_start = time.time() + try: + retry_ok, retry_fail, retry_pending = process_retries(cur) + + cur.execute(""" + SELECT * FROM ( + SELECT DISTINCT ON (packet_hash) + packet_hash, time, + payload->'decrypted'->>'sender' AS sender, + payload->'decrypted'->>'message' AS message + FROM packets + WHERE packet_type = 5 + AND payload->'decrypted'->>'message' IS NOT NULL + AND time > %s::timestamptz - make_interval(secs => %s) + AND time <= now() + AND network = ANY(%s::text[]) + AND visibility_ok IS TRUE + AND is_private IS NOT TRUE + AND EXISTS ( + SELECT 1 + FROM packet_visibility_materialization_state cached_visibility + JOIN public_visibility_state current_visibility + ON current_visibility.singleton = cached_visibility.singleton + WHERE cached_visibility.singleton = TRUE + AND cached_visibility.visibility_generation = current_visibility.generation + ) + ORDER BY packet_hash, time DESC + ) x ORDER BY time ASC LIMIT %s + """, (cursor, OVERLAP_S, NETWORKS, BATCH_LIMIT)) + rows = cur.fetchall() + new = errs = 0 + max_t = cursor + for ph, ts, sender, message in rows: + if ts > max_t: + max_t = ts + cur.execute("SELECT 1 FROM message_tags WHERE packet_hash=%s", (ph,)) + if cur.fetchone(): + continue + state = f"Message from node '{sender or 'unknown'}': \"{(message or '')[:4000]}\"" + ans, ms, err = None, None, None + for attempt in (1, 2, 3): + try: + resp, ms = api_tag(state) + ans = resp.get('answers') + break + except urllib.error.HTTPError as e: + if e.code == 429: + cur.execute("UPDATE tagger_state SET http_429=http_429+1 WHERE id=1") + time.sleep(min(CALL_FLOOR_S * (2 ** attempt), 30)) + continue + err = f'http {e.code}: {e.read().decode()[:160]}' + time.sleep(2 * attempt) + except Exception as ex: # noqa: BLE001 + err = str(ex)[:200] + time.sleep(2 * attempt) + if ans: + tags, conf = flatten(ans) + cur.execute("""INSERT INTO message_tags + (packet_hash, packet_time, tagged_at, model, latency_ms, tags, confidence) + VALUES (%s, %s, now(), %s, %s, %s, %s) + ON CONFLICT (packet_hash) DO NOTHING""", + (ph, ts, MODEL, int(ms), json.dumps(tags), json.dumps(conf))) + new += 1 + lat_hist.append(ms) + lat_hist = lat_hist[-50:] + else: + errs += 1 + cur.execute("UPDATE tagger_state SET errors=errors+1, last_error=%s WHERE id=1", (err,)) + # Persist for retry instead of letting the cursor move past it. + cur.execute("""INSERT INTO tagger_retry (packet_hash, packet_time, attempts, last_error) + VALUES (%s, %s, 1, %s) + ON CONFLICT (packet_hash) DO UPDATE + SET attempts=tagger_retry.attempts+1, last_error=EXCLUDED.last_error""", + (ph, ts, err)) + time.sleep(CALL_FLOOR_S) + if max_t and max_t != cursor: + cur.execute("UPDATE tagger_state SET cursor_time=%s WHERE id=1", (max_t,)) + cursor = max_t + avg = (sum(lat_hist) / len(lat_hist)) if lat_hist else None + cur.execute("""UPDATE tagger_state SET seen=seen+%s, tagged=tagged+%s, + avg_latency_ms_50=%s, updated_at=now() WHERE id=1""", + (new + errs, new, avg)) + if new or errs or retry_ok or retry_fail or (time.time() - last_log) > 120: + log(f"batch={len(rows)} new={new} errs={errs} retry_ok={retry_ok} " + f"retry_pending={retry_pending} avg50={avg and round(avg)}ms cursor={cursor.isoformat()}") + last_log = time.time() + if new + errs == 0 and retry_ok + retry_fail == 0: + time.sleep(POLL_S) + except Exception as ex: # noqa: BLE001 + log(f"loop error: {ex}") + try: + conn.rollback() + except Exception: # noqa: BLE001 + pass + time.sleep(POLL_S) + try: + conn = psycopg2.connect(DATABASE_URL) + conn.autocommit = True + cur = conn.cursor() + except Exception as ex2: # noqa: BLE001 + log(f"reconnect failed: {ex2}") + time.sleep(15) + _ = loop_start + log('shutting down') + + +if __name__ == '__main__': + main() diff --git a/vacuum-compressed-chunks.sh b/vacuum-compressed-chunks.sh index e01f1b1..abec917 100755 --- a/vacuum-compressed-chunks.sh +++ b/vacuum-compressed-chunks.sh @@ -59,8 +59,18 @@ if [ -n "$chunk" ]; then fi cd "$repo_dir" -docker compose config -q -psql_command=(docker compose exec -T timescaledb psql -X -v ON_ERROR_STOP=1 -U "${POSTGRES_USER:-meshcore}" -d "$database") +# Resolve the compose project that owns the timescaledb service. The test harness +# drives the script with COMPOSE_FILE/COMPOSE_PROJECT_NAME set (its own fixture); +# in the two-project production layout the database lives in the separate +# meshcore-infra project (mirrors scripts/backup.sh). +if [ -z "${COMPOSE_FILE:-}" ]; then + infra_dir="${MESHCORE_INFRA_DIR:-${repo_dir}/../meshcore-infra}" + compose_base=(docker compose --project-directory "$infra_dir" -f "$infra_dir/docker-compose.yml" --project-name "${MESHCORE_INFRA_PROJECT_NAME:-meshcore-infra}") +else + compose_base=(docker compose) +fi +"${compose_base[@]}" config -q +psql_command=("${compose_base[@]}" exec -T timescaledb psql -X -v ON_ERROR_STOP=1 -U "${POSTGRES_USER:-meshcore}" -d "$database") database_exists="$("${psql_command[@]}" -Atc \ "SELECT 1 FROM pg_database WHERE datname = current_database() AND NOT datistemplate")" @@ -128,7 +138,7 @@ case "$action" in ;; esac available_bytes="$( - docker compose exec -T timescaledb \ + "${compose_base[@]}" exec -T timescaledb \ df -PB1 /var/lib/postgresql/data | awk 'NR==2 {print $4}' )"