mirror of
https://github.com/gadgethd/ukmesh.git
synced 2026-10-23 14:22:47 +00:00
feat(tags): live message tags for feed + app
- tagger-worker: persistent retry queue (tagger_retry) so API outages cannot drop tags; overlap 20s->120s; only publicly-visible in-scope messages are sent to the API (visibility + network filter) - backend: tags joined into /feed/messages, /packets/recent, /packets/:hash and the WS initial state; new GET /tags/recent for live enrichment; contract + openapi entries - frontend: tag chips under the row meta line (feed + app), live 15s tag poll, Message tags section in the packet detail panel
This commit is contained in:
gadgethd committed
1 parent
4bf39aea94
commit
1d86e043a5
19 files changed
+1380
-21
No files matched your search
@@ -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;
|
||||
});
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -51,6 +51,7 @@ type GetChannelMessageHistoryFn = (
|
||||
network?: string,
|
||||
observer?: string,
|
||||
) => Promise<unknown>;
|
||||
type GetRecentMessageTagsFn = (limit: number, network?: string) => Promise<unknown>;
|
||||
|
||||
type MiscRouteDeps = {
|
||||
query: QueryFn;
|
||||
@@ -58,6 +59,7 @@ type MiscRouteDeps = {
|
||||
getRecentPacketEvents: GetRecentPacketEventsFn;
|
||||
getPacketDetail: GetPacketDetailFn;
|
||||
getChannelMessageHistory: GetChannelMessageHistoryFn;
|
||||
getRecentMessageTags: GetRecentMessageTagsFn;
|
||||
getPublicVisibilityGeneration: () => Promise<number>;
|
||||
packetDetailLimiter: ReturnType<typeof import('express-rate-limit').rateLimit>;
|
||||
};
|
||||
@@ -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',
|
||||
|
||||
+101
-15
@@ -149,6 +149,11 @@ async function queryPool<T extends pg.QueryResultRow = pg.QueryResultRow>(
|
||||
}
|
||||
}
|
||||
|
||||
export type DatabaseQueryFn = <T extends pg.QueryResultRow = pg.QueryResultRow>(
|
||||
text: string,
|
||||
params?: unknown[],
|
||||
) => Promise<{ rows: T[] }>;
|
||||
|
||||
export async function query<T extends pg.QueryResultRow = pg.QueryResultRow>(
|
||||
text: string,
|
||||
params?: unknown[],
|
||||
@@ -157,6 +162,28 @@ export async function query<T extends pg.QueryResultRow = pg.QueryResultRow>(
|
||||
return queryPool<T>(pool, 'oltp', text, params, signal);
|
||||
}
|
||||
|
||||
export async function withTransaction<T>(work: (transactionQuery: DatabaseQueryFn) => Promise<T>): Promise<T> {
|
||||
const client = await pool.connect();
|
||||
const transactionQuery: DatabaseQueryFn = async <R extends pg.QueryResultRow = pg.QueryResultRow>(
|
||||
text: string,
|
||||
params?: unknown[],
|
||||
) => {
|
||||
const result = await client.query<R>(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<T extends pg.QueryResultRow = pg.QueryResultRow>(
|
||||
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<string, unknown> | null) ?? null,
|
||||
tagConfidence: (row.tag_confidence as Record<string, unknown> | null) ?? null,
|
||||
observations: observations.rows.map((r) => ({
|
||||
rxNodeId: r.rx_node_id as string | null,
|
||||
iata: r.iata as string | null,
|
||||
|
||||
@@ -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'],
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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';
|
||||
|
||||
|
||||
@@ -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<PacketFeedItemProps> = React.memo(({
|
||||
</span>
|
||||
</>}
|
||||
</span>
|
||||
{messageTagKind(p) && (
|
||||
<span className="packet-item__tags">
|
||||
<span className="packet-item__tag msg-tag-chip" title={messageTagTitle(p)}>{messageTagKind(p)}</span>
|
||||
</span>
|
||||
)}
|
||||
</div>
|
||||
<div className="packet-item__actions">
|
||||
<button
|
||||
|
||||
@@ -4,6 +4,52 @@ import { canonicalNodeId, canonicalOptionalNodeId } from '../utils/nodeIds.js';
|
||||
export const FEED_MAX_PACKETS = 50;
|
||||
export const FEED_MAX_MESSAGES = 200;
|
||||
|
||||
/** Message tags produced by the tagger-worker (TypeSafe/Jev). Flags are 0..1
|
||||
* confidences; kind/topic/speaker are choices with their own confidences. */
|
||||
export type MessageTags = {
|
||||
kind?: string;
|
||||
topic?: string;
|
||||
speaker?: string;
|
||||
safety?: number;
|
||||
directed?: number;
|
||||
mentions_location?: number;
|
||||
};
|
||||
|
||||
export type TagConfidence = {
|
||||
kind?: number;
|
||||
topic?: number;
|
||||
speaker?: number;
|
||||
};
|
||||
|
||||
/** One row of GET /api/tags/recent (live feed enrichment by packet hash). */
|
||||
export type TaggedMessage = {
|
||||
packet_hash: string;
|
||||
tags: MessageTags;
|
||||
confidence?: TagConfidence | null;
|
||||
tagged_at?: string;
|
||||
};
|
||||
|
||||
export function messageTagKind(packet: { tags?: MessageTags }): string | null {
|
||||
const kind = packet.tags?.kind;
|
||||
return typeof kind === 'string' && kind.trim() ? kind.trim() : null;
|
||||
}
|
||||
|
||||
export function messageTagTitle(packet: { tags?: MessageTags; tag_confidence?: TagConfidence }): string {
|
||||
const tags = packet.tags ?? {};
|
||||
const parts: string[] = [];
|
||||
if (tags.kind) parts.push(`kind: ${tags.kind}`);
|
||||
if (tags.topic) parts.push(`topic: ${tags.topic}`);
|
||||
if (tags.speaker) parts.push(`speaker: ${tags.speaker}`);
|
||||
const flags: string[] = [];
|
||||
if ((tags.safety ?? 0) >= 0.5) flags.push('safety');
|
||||
if ((tags.directed ?? 0) >= 0.5) flags.push('directed');
|
||||
if ((tags.mentions_location ?? 0) >= 0.5) flags.push('location');
|
||||
if (flags.length > 0) parts.push(`flags: ${flags.join(', ')}`);
|
||||
const conf = packet.tag_confidence?.kind;
|
||||
if (typeof conf === 'number') parts.push(`kind confidence ${(conf * 100).toFixed(0)}%`);
|
||||
return parts.join(' · ');
|
||||
}
|
||||
|
||||
export type RecentPacketRow = {
|
||||
time: string;
|
||||
packet_hash: string;
|
||||
@@ -22,6 +68,8 @@ export type RecentPacketRow = {
|
||||
path_hashes?: string[] | null;
|
||||
rx_count?: number | null;
|
||||
tx_count?: number | null;
|
||||
tags?: MessageTags | null;
|
||||
tag_confidence?: TagConfidence | null;
|
||||
};
|
||||
|
||||
export function normalizeIatas(values: Array<string | null | undefined>): string[] {
|
||||
@@ -102,6 +150,9 @@ export function mergeAggregatedPacket(current: AggregatedPacket, next: Aggregate
|
||||
txCount: Math.max(current.txCount, next.txCount),
|
||||
ts: Math.max(current.ts, next.ts),
|
||||
advertCount: Math.max(current.advertCount ?? 0, next.advertCount ?? 0) || undefined,
|
||||
tags: next.tags ?? current.tags,
|
||||
tagConfidence: next.tagConfidence ?? current.tagConfidence,
|
||||
taggedAt: next.taggedAt ?? current.taggedAt,
|
||||
};
|
||||
|
||||
if (packetInfoScore(mergedCandidate) >= packetInfoScore(current)) return mergedCandidate;
|
||||
@@ -145,6 +196,8 @@ export function mapRecentRows(rows: RecentPacketRow[]): AggregatedPacket[] {
|
||||
txCount: Number(row.tx_count ?? 0),
|
||||
ts: new Date(row.time).getTime(),
|
||||
advertCount: row.advert_count ?? undefined,
|
||||
tags: row.tags ?? undefined,
|
||||
tagConfidence: row.tag_confidence ?? undefined,
|
||||
};
|
||||
const current = mapped.get(packetHash);
|
||||
mapped.set(packetHash, current ? mergeAggregatedPacket(current, next) : next);
|
||||
@@ -182,6 +235,8 @@ export function mapMessageRows(rows: RecentPacketRow[]): AggregatedPacket[] {
|
||||
txCount: Number(row.tx_count ?? 0),
|
||||
ts: new Date(row.time).getTime(),
|
||||
advertCount: row.advert_count ?? undefined,
|
||||
tags: row.tags ?? undefined,
|
||||
tagConfidence: row.tag_confidence ?? undefined,
|
||||
};
|
||||
const current = mapped.get(packetHash);
|
||||
mapped.set(packetHash, current ? mergeAggregatedPacket(current, next) : next);
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
import { useEffect, useRef } from 'react';
|
||||
import { fetchJson, uncachedEndpoint, withScopeParams } from '../utils/api.js';
|
||||
import { nodeStore } from './useNodes.js';
|
||||
import type { TaggedMessage } from './packetFeed.js';
|
||||
|
||||
const TAGS_POLL_MS = 15_000;
|
||||
|
||||
type TagScope = { network?: string; observer?: string };
|
||||
|
||||
/** Polls GET /api/tags/recent and patches tags onto store rows by packet hash.
|
||||
* Tags are written by tagger-worker ~2s after a packet arrives, so live WS rows
|
||||
* start untagged and get their chips on the next poll. Best-effort: failures
|
||||
* are ignored (the feed keeps working untagged). */
|
||||
export function useMessageTags(scope: TagScope, scopeKey: string): void {
|
||||
const scopeRef = useRef(scope);
|
||||
scopeRef.current = scope;
|
||||
|
||||
useEffect(() => {
|
||||
let cancelled = false;
|
||||
let timer: number | null = null;
|
||||
const schedule = () => {
|
||||
if (cancelled) return;
|
||||
timer = window.setTimeout(() => void tick(), TAGS_POLL_MS);
|
||||
};
|
||||
const tick = async () => {
|
||||
if (cancelled) return;
|
||||
if (typeof document !== 'undefined' && document.visibilityState === 'hidden') {
|
||||
schedule();
|
||||
return;
|
||||
}
|
||||
try {
|
||||
const rows = await fetchJson<TaggedMessage[]>(
|
||||
uncachedEndpoint(withScopeParams('/api/tags/recent?limit=300', scopeRef.current)),
|
||||
{ cache: 'no-store' },
|
||||
{ timeoutMs: 15_000, maxBytes: 2 * 1024 * 1024 },
|
||||
);
|
||||
if (!cancelled && Array.isArray(rows)) nodeStore.applyMessageTags(rows);
|
||||
} catch {
|
||||
/* enrichment is best-effort */
|
||||
}
|
||||
schedule();
|
||||
};
|
||||
void tick();
|
||||
return () => {
|
||||
cancelled = true;
|
||||
if (timer !== null) window.clearTimeout(timer);
|
||||
};
|
||||
}, [scopeKey]);
|
||||
}
|
||||
@@ -8,6 +8,8 @@ import {
|
||||
mergePackets,
|
||||
packetInfoScore,
|
||||
type RecentPacketRow,
|
||||
type MessageTags,
|
||||
type TagConfidence,
|
||||
FEED_MAX_MESSAGES,
|
||||
FEED_MAX_PACKETS,
|
||||
} from './packetFeed.js';
|
||||
@@ -69,6 +71,9 @@ export interface AggregatedPacket {
|
||||
txCount: number;
|
||||
ts: number;
|
||||
advertCount?: number;
|
||||
tags?: MessageTags;
|
||||
tagConfidence?: TagConfidence;
|
||||
taggedAt?: string;
|
||||
}
|
||||
|
||||
export interface PacketArc {
|
||||
@@ -270,6 +275,29 @@ function handleInitialState(
|
||||
});
|
||||
}
|
||||
|
||||
/** Patch tags onto cached rows as the tagger finishes them (tags are produced a
|
||||
* few seconds after arrival, so live WS rows start untagged). Idempotent:
|
||||
* rows already carrying the same tagged_at are left untouched. */
|
||||
function applyMessageTags(entries: Array<{ packet_hash?: string; tags?: MessageTags; confidence?: TagConfidence | null; tagged_at?: string }>): void {
|
||||
const byHash = new Map<string, { tags: MessageTags; confidence?: TagConfidence | null; tagged_at?: string }>();
|
||||
for (const entry of entries) {
|
||||
const hash = String(entry.packet_hash ?? '').trim().toUpperCase();
|
||||
if (hash && entry.tags) byHash.set(hash, { tags: entry.tags, confidence: entry.confidence, tagged_at: entry.tagged_at });
|
||||
}
|
||||
if (byHash.size === 0) return;
|
||||
let changed = false;
|
||||
const patchList = (list: AggregatedPacket[]): AggregatedPacket[] => list.map((packet) => {
|
||||
const entry = byHash.get(packet.packetHash);
|
||||
if (!entry) return packet;
|
||||
if (packet.taggedAt && entry.tagged_at && packet.taggedAt === entry.tagged_at) return packet;
|
||||
changed = true;
|
||||
return { ...packet, tags: entry.tags, tagConfidence: entry.confidence ?? undefined, taggedAt: entry.tagged_at };
|
||||
});
|
||||
const packets = patchList(state.packets);
|
||||
const messages = patchList(state.messages);
|
||||
if (changed) setState({ ...state, packets, messages });
|
||||
}
|
||||
|
||||
function replaceRecentPackets(rows: RecentPacketRow[], epoch?: number) {
|
||||
if (!acceptsEpoch(epoch)) return;
|
||||
const mapped = mapRecentRows(rows);
|
||||
@@ -519,6 +547,7 @@ export const nodeStore = {
|
||||
handleNodeUpsertBatch,
|
||||
setArcCollectionEnabled,
|
||||
pruneExpiredArcs,
|
||||
applyMessageTags,
|
||||
};
|
||||
|
||||
export function useNodeMap(): Map<string, MeshNode> {
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { useCallback, useEffect, useState } from 'react';
|
||||
import type { LazyPathResult } from '../pages/ukmesh/PacketDetailPanel.js';
|
||||
import type { MultiObserverBetaResponse } from './packetPathOverlayUtils.js';
|
||||
import type { MessageTags, TagConfidence } from './packetFeed.js';
|
||||
import { useRuntimeFeatures } from '../config/runtimeFeatures.js';
|
||||
import { ApiResponseError, fetchJson, withScopeParams } from '../utils/api.js';
|
||||
import { ScopedCache } from '../utils/scopedCache.js';
|
||||
@@ -21,6 +22,8 @@ export type PacketDetail = {
|
||||
pathHashes: string[] | null;
|
||||
pathHashSizeBytes: number | null;
|
||||
rawHex: string | null;
|
||||
tags?: MessageTags | null;
|
||||
tagConfidence?: TagConfidence | null;
|
||||
observations: Array<{ rxNodeId: string | null; iata: string | null; time: string; rssi: number | null; snr: number | null; hopCount: number | null }>;
|
||||
};
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@ import { LoadingIndicator } from '../../components/LoadingIndicator.js';
|
||||
import type { MeshNode } from '../../hooks/useNodes.js';
|
||||
import type { FeedPacket } from './UKFeedPage.js';
|
||||
import { usePacketDetailData, type RadioState } from '../../hooks/usePacketDetailData.js';
|
||||
import type { MessageTags, TagConfidence } from '../../hooks/packetFeed.js';
|
||||
import {
|
||||
PathMap,
|
||||
type LazyPath,
|
||||
@@ -335,6 +336,32 @@ export const PacketDetailPanel: React.FC<{
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{/* 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 (
|
||||
<div className="feed-detail__section">
|
||||
<div className="feed-detail__section-title">
|
||||
Message tags{typeof conf?.kind === 'number' ? ` · kind ${(conf.kind * 100).toFixed(0)}%` : ''}
|
||||
</div>
|
||||
<div className="feed-detail__tags">
|
||||
{chips.map((chip) => (
|
||||
<span key={chip} className="msg-tag-chip">{chip}</span>
|
||||
))}
|
||||
</div>
|
||||
</div>
|
||||
);
|
||||
})()}
|
||||
|
||||
{/* Observer table */}
|
||||
{detail?.observations && detail.observations.length > 0 && (
|
||||
<div className="feed-detail__section">
|
||||
|
||||
@@ -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 = () => {
|
||||
<span className="uk-feed-packet-row__hash dev-status-mono">{packet.packet_hash}</span>
|
||||
<span className="uk-feed-packet-row__observer">{observerDisplay}</span>
|
||||
</div>
|
||||
{messageTagKind(packet) && (
|
||||
<div className="uk-feed-packet-row__tags">
|
||||
<span className="msg-tag-chip" title={messageTagTitle(packet)}>
|
||||
{messageTagKind(packet)}
|
||||
</span>
|
||||
</div>
|
||||
)}
|
||||
<p className="uk-feed-packet-row__summary">{packetSummary(packet, nodeMap)}</p>
|
||||
</article>
|
||||
</React.Fragment>
|
||||
@@ -689,6 +704,11 @@ export const UKFeedPage: React.FC = () => {
|
||||
{selectedPacket.hop_count} hop{selectedPacket.hop_count !== 1 ? 's' : ''}
|
||||
</span>
|
||||
)}
|
||||
{messageTagKind(selectedPacket) && (
|
||||
<span className="msg-tag-chip" title={messageTagTitle(selectedPacket)}>
|
||||
{messageTagKind(selectedPacket)}
|
||||
</span>
|
||||
)}
|
||||
</div>
|
||||
<p className="uk-feed-stats__selected-summary">{packetSummary(selectedPacket, nodeMap)}</p>
|
||||
<div className="uk-feed-stats__actions">
|
||||
@@ -748,6 +768,11 @@ export const UKFeedPage: React.FC = () => {
|
||||
{selectedPacket.hop_count} hop{selectedPacket.hop_count !== 1 ? 's' : ''}
|
||||
</span>
|
||||
)}
|
||||
{messageTagKind(selectedPacket) && (
|
||||
<span className="msg-tag-chip" title={messageTagTitle(selectedPacket)}>
|
||||
{messageTagKind(selectedPacket)}
|
||||
</span>
|
||||
)}
|
||||
<button type="button" className="uk-feed-stats__close" onClick={() => setSelectedPacketHash(null)} aria-label="Clear selected packet">✕</button>
|
||||
</div>
|
||||
<p className="uk-feed-stats__selected-summary">{packetSummary(selectedPacket, nodeMap)}</p>
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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<number, string> = {
|
||||
@@ -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, MeshNode>): string[] {
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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()
|
||||
@@ -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}'
|
||||
)"
|
||||
|
||||
|
||||
Reference in new issue
Block a user