adam/Remote-Terminal-for-MeshCore
1
0
Fork 0
mirror of https://github.com/jkingsman/Remote-Terminal-for-MeshCore.git synced 2026-08-08 01:33:01 +02:00
Code Issues Packages Projects Releases Wiki Activity

Contact info pane

Browse Source
This commit is contained in:
Jack Kingsman
2026-02-27 13:38:53 -08:00
parent 24166e92e8
commit b91b2d5d7b
28 changed files with 1624 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/AGENTS.md
+7 -2
View File
@@ -114,7 +114,10 @@ app/
### Contacts
- `GET /contacts`
- `GET /contacts/repeaters/advert-paths` — recent advert paths for all contacts
- `GET /contacts/{public_key}`
- `GET /contacts/{public_key}/detail` — comprehensive contact profile (stats, name history, paths, nearest repeaters)
- `GET /contacts/{public_key}/advert-paths` — recent advert paths for one contact
- `POST /contacts`
- `DELETE /contacts/{public_key}`
- `POST /contacts/sync`
@@ -176,10 +179,12 @@ Client sends `"ping"` text; server replies `{"type":"pong"}`.
## Data Model Notes
Main tables:
- `contacts`
- `contacts` (includes `first_seen` for contact age tracking)
- `channels`
- `messages`
- `messages` (includes `sender_name`, `sender_key` for per-contact channel message attribution)
- `raw_packets`
- `contact_advert_paths` (recent unique advertisement paths per contact)
- `contact_name_history` (tracks name changes over time)
- `app_settings`
`app_settings` fields in active model:
app/database.py
+24 -8
View File
@@ -20,7 +20,8 @@ CREATE TABLE IF NOT EXISTS contacts (
lon REAL,
last_seen INTEGER,
on_radio INTEGER DEFAULT 0,
last_contacted INTEGER
last_contacted INTEGER,
first_seen INTEGER
);
CREATE TABLE IF NOT EXISTS channels (
@@ -41,7 +42,9 @@ CREATE TABLE IF NOT EXISTS messages (
txt_type INTEGER DEFAULT 0,
signature TEXT,
outgoing INTEGER DEFAULT 0,
acked INTEGER DEFAULT 0
acked INTEGER DEFAULT 0,
sender_name TEXT,
sender_key TEXT
-- Deduplication: identical text + timestamp in the same conversation is treated as a
-- mesh echo/repeat. Second-precision timestamps mean two intentional identical messages
-- within the same second would collide, but this is not feasible in practice — LoRa
@@ -59,16 +62,26 @@ CREATE TABLE IF NOT EXISTS raw_packets (
FOREIGN KEY (message_id) REFERENCES messages(id)
);
CREATE TABLE IF NOT EXISTS repeater_advert_paths (
CREATE TABLE IF NOT EXISTS contact_advert_paths (
id INTEGER PRIMARY KEY AUTOINCREMENT,
repeater_key TEXT NOT NULL,
public_key TEXT NOT NULL,
path_hex TEXT NOT NULL,
path_len INTEGER NOT NULL,
first_seen INTEGER NOT NULL,
last_seen INTEGER NOT NULL,
heard_count INTEGER NOT NULL DEFAULT 1,
UNIQUE(repeater_key, path_hex),
FOREIGN KEY (repeater_key) REFERENCES contacts(public_key)
UNIQUE(public_key, path_hex),
FOREIGN KEY (public_key) REFERENCES contacts(public_key)
);
CREATE TABLE IF NOT EXISTS contact_name_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
public_key TEXT NOT NULL,
name TEXT NOT NULL,
first_seen INTEGER NOT NULL,
last_seen INTEGER NOT NULL,
UNIQUE(public_key, name),
FOREIGN KEY (public_key) REFERENCES contacts(public_key)
);
CREATE INDEX IF NOT EXISTS idx_messages_conversation ON messages(type, conversation_key);
@@ -78,8 +91,11 @@ CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_dedup_null_safe
CREATE INDEX IF NOT EXISTS idx_raw_packets_message_id ON raw_packets(message_id);
CREATE UNIQUE INDEX IF NOT EXISTS idx_raw_packets_payload_hash ON raw_packets(payload_hash);
CREATE INDEX IF NOT EXISTS idx_contacts_on_radio ON contacts(on_radio);
CREATE INDEX IF NOT EXISTS idx_repeater_advert_paths_recent
ON repeater_advert_paths(repeater_key, last_seen DESC);
-- idx_messages_sender_key is created by migration 25 (after adding the sender_key column)
CREATE INDEX IF NOT EXISTS idx_contact_advert_paths_recent
ON contact_advert_paths(public_key, last_seen DESC);
CREATE INDEX IF NOT EXISTS idx_contact_name_history_key
ON contact_name_history(public_key, last_seen DESC);
"""
app/event_handlers.py
+13 -1
View File
@@ -7,7 +7,12 @@ from meshcore import EventType
from app.models import CONTACT_TYPE_REPEATER, Contact
from app.packet_processor import process_raw_packet
from app.repository import AmbiguousPublicKeyPrefixError, ContactRepository, MessageRepository
from app.repository import (
AmbiguousPublicKeyPrefixError,
ContactNameHistoryRepository,
ContactRepository,
MessageRepository,
)
from app.websocket import broadcast_event
if TYPE_CHECKING:
@@ -251,6 +256,13 @@ async def on_new_contact(event: "Event") -> None:
}
await ContactRepository.upsert(contact_data)
# Record name history if contact has a name
adv_name = payload.get("adv_name")
if adv_name:
await ContactNameHistoryRepository.record_name(
public_key.lower(), adv_name, int(time.time())
)
# Read back from DB so the broadcast includes all fields (last_contacted,
# last_read_at, etc.) matching the REST Contact shape exactly.
db_contact = await ContactRepository.get_by_key(public_key)
app/migrations.py
+364
View File
@@ -191,6 +191,41 @@ async def run_migrations(conn: aiosqlite.Connection) -> int:
await set_version(conn, 22)
applied += 1
# Migration 23: Add first_seen column to contacts
if version < 23:
logger.info("Applying migration 23: add first_seen column to contacts")
await _migrate_023_add_first_seen(conn)
await set_version(conn, 23)
applied += 1
# Migration 24: Create contact_name_history table
if version < 24:
logger.info("Applying migration 24: create contact_name_history table")
await _migrate_024_create_contact_name_history(conn)
await set_version(conn, 24)
applied += 1
# Migration 25: Add sender_name and sender_key to messages
if version < 25:
logger.info("Applying migration 25: add sender_name and sender_key to messages")
await _migrate_025_add_sender_columns(conn)
await set_version(conn, 25)
applied += 1
# Migration 26: Rename repeater_advert_paths to contact_advert_paths
if version < 26:
logger.info("Applying migration 26: rename repeater_advert_paths to contact_advert_paths")
await _migrate_026_rename_advert_paths_table(conn)
await set_version(conn, 26)
applied += 1
# Migration 27: Backfill first_seen from advert paths
if version < 27:
logger.info("Applying migration 27: backfill first_seen from advert paths")
await _migrate_027_backfill_first_seen_from_advert_paths(conn)
await set_version(conn, 27)
applied += 1
if applied > 0:
logger.info(
"Applied %d migration(s), schema now at version %d", applied, await get_version(conn)
@@ -1318,3 +1353,332 @@ async def _migrate_022_add_repeater_advert_paths(conn: aiosqlite.Connection) ->
)
await conn.commit()
logger.debug("Ensured repeater_advert_paths table and indexes exist")
async def _migrate_023_add_first_seen(conn: aiosqlite.Connection) -> None:
"""
Add first_seen column to contacts table.
Backfill strategy:
1. Set first_seen = last_seen for all contacts (baseline).
2. For contacts with PRIV messages, set first_seen = MIN(messages.received_at)
if that timestamp is earlier.
"""
# Guard: skip if contacts table doesn't exist (e.g. partial test schemas)
cursor = await conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='contacts'"
)
if not await cursor.fetchone():
return
try:
await conn.execute("ALTER TABLE contacts ADD COLUMN first_seen INTEGER")
logger.debug("Added first_seen to contacts table")
except aiosqlite.OperationalError as e:
if "duplicate column name" in str(e).lower():
logger.debug("contacts.first_seen already exists, skipping")
else:
raise
# Baseline: set first_seen = last_seen for all contacts
# Check if last_seen column exists (should in production, may not in minimal test schemas)
cursor = await conn.execute("PRAGMA table_info(contacts)")
columns = {row[1] for row in await cursor.fetchall()}
if "last_seen" in columns:
await conn.execute("UPDATE contacts SET first_seen = last_seen WHERE first_seen IS NULL")
# Refine: for contacts with PRIV messages, use earliest message timestamp if earlier
cursor = await conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='messages'"
)
if await cursor.fetchone():
await conn.execute(
"""
UPDATE contacts SET first_seen = (
SELECT MIN(m.received_at) FROM messages m
WHERE m.type = 'PRIV' AND m.conversation_key = contacts.public_key
)
WHERE EXISTS (
SELECT 1 FROM messages m
WHERE m.type = 'PRIV' AND m.conversation_key = contacts.public_key
AND m.received_at < COALESCE(contacts.first_seen, 9999999999)
)
"""
)
await conn.commit()
logger.debug("Added and backfilled first_seen column")
async def _migrate_024_create_contact_name_history(conn: aiosqlite.Connection) -> None:
"""
Create contact_name_history table and seed with current contact names.
"""
await conn.execute(
"""
CREATE TABLE IF NOT EXISTS contact_name_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
public_key TEXT NOT NULL,
name TEXT NOT NULL,
first_seen INTEGER NOT NULL,
last_seen INTEGER NOT NULL,
UNIQUE(public_key, name),
FOREIGN KEY (public_key) REFERENCES contacts(public_key)
)
"""
)
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_contact_name_history_key "
"ON contact_name_history(public_key, last_seen DESC)"
)
# Seed: one row per contact from current data (skip if contacts table doesn't exist
# or lacks needed columns)
cursor = await conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='contacts'"
)
if await cursor.fetchone():
cursor = await conn.execute("PRAGMA table_info(contacts)")
cols = {row[1] for row in await cursor.fetchall()}
if "name" in cols and "public_key" in cols:
first_seen_expr = "first_seen" if "first_seen" in cols else "0"
last_seen_expr = "last_seen" if "last_seen" in cols else "0"
await conn.execute(
f"""
INSERT OR IGNORE INTO contact_name_history (public_key, name, first_seen, last_seen)
SELECT public_key, name,
COALESCE({first_seen_expr}, {last_seen_expr}, 0),
COALESCE({last_seen_expr}, 0)
FROM contacts
WHERE name IS NOT NULL AND name != ''
"""
)
await conn.commit()
logger.debug("Created contact_name_history table and seeded from contacts")
async def _migrate_025_add_sender_columns(conn: aiosqlite.Connection) -> None:
"""
Add sender_name and sender_key columns to messages table.
Backfill:
- sender_name for CHAN messages: extract from "Name: message" format
- sender_key for CHAN messages: match name to contact (skip ambiguous)
- sender_key for incoming PRIV messages: set to conversation_key
"""
# Guard: skip if messages table doesn't exist
cursor = await conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='messages'"
)
if not await cursor.fetchone():
return
for column in ["sender_name", "sender_key"]:
try:
await conn.execute(f"ALTER TABLE messages ADD COLUMN {column} TEXT")
logger.debug("Added %s to messages table", column)
except aiosqlite.OperationalError as e:
if "duplicate column name" in str(e).lower():
logger.debug("messages.%s already exists, skipping", column)
else:
raise
# Check which columns the messages table has (may be minimal in test environments)
cursor = await conn.execute("PRAGMA table_info(messages)")
msg_cols = {row[1] for row in await cursor.fetchall()}
# Only backfill if the required columns exist
if "type" in msg_cols and "text" in msg_cols:
# Count messages to backfill for progress reporting
cursor = await conn.execute(
"SELECT COUNT(*) FROM messages WHERE type = 'CHAN' AND sender_name IS NULL"
)
row = await cursor.fetchone()
chan_count = row[0] if row else 0
if chan_count > 0:
logger.info("Backfilling sender_name for %d channel messages...", chan_count)
# Backfill sender_name for CHAN messages from "Name: message" format
# Only extract if colon position is valid (> 1 and < 51, i.e. name is 1-50 chars)
cursor = await conn.execute(
"""
UPDATE messages SET sender_name = SUBSTR(text, 1, INSTR(text, ': ') - 1)
WHERE type = 'CHAN' AND sender_name IS NULL
AND INSTR(text, ': ') > 1 AND INSTR(text, ': ') < 52
"""
)
if cursor.rowcount > 0:
logger.info("Backfilled sender_name for %d channel messages", cursor.rowcount)
# Backfill sender_key for incoming PRIV messages
if "outgoing" in msg_cols and "conversation_key" in msg_cols:
cursor = await conn.execute(
"""
UPDATE messages SET sender_key = conversation_key
WHERE type = 'PRIV' AND outgoing = 0 AND sender_key IS NULL
"""
)
if cursor.rowcount > 0:
logger.info("Backfilled sender_key for %d DM messages", cursor.rowcount)
# Backfill sender_key for CHAN messages: match sender_name to contacts
# Build name->key map, skip ambiguous names (multiple contacts with same name)
cursor = await conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='contacts'"
)
if await cursor.fetchone():
cursor = await conn.execute(
"SELECT public_key, name FROM contacts WHERE name IS NOT NULL AND name != ''"
)
rows = await cursor.fetchall()
name_to_keys: dict[str, list[str]] = {}
for row in rows:
name = row["name"]
key = row["public_key"]
if name not in name_to_keys:
name_to_keys[name] = []
name_to_keys[name].append(key)
# Only use unambiguous names (single contact per name)
unambiguous = {n: ks[0] for n, ks in name_to_keys.items() if len(ks) == 1}
if unambiguous:
logger.info(
"Matching sender_key for %d unique contact names...",
len(unambiguous),
)
# Use a temp table for a single bulk UPDATE instead of N individual queries
await conn.execute(
"CREATE TEMP TABLE _name_key_map (name TEXT PRIMARY KEY, public_key TEXT NOT NULL)"
)
await conn.executemany(
"INSERT INTO _name_key_map (name, public_key) VALUES (?, ?)",
list(unambiguous.items()),
)
cursor = await conn.execute(
"""
UPDATE messages SET sender_key = (
SELECT public_key FROM _name_key_map WHERE _name_key_map.name = messages.sender_name
)
WHERE type = 'CHAN' AND sender_key IS NULL
AND sender_name IN (SELECT name FROM _name_key_map)
"""
)
updated = cursor.rowcount
await conn.execute("DROP TABLE _name_key_map")
if updated > 0:
logger.info("Backfilled sender_key for %d channel messages", updated)
# Create index on sender_key for per-contact channel message counts
await conn.execute("CREATE INDEX IF NOT EXISTS idx_messages_sender_key ON messages(sender_key)")
await conn.commit()
logger.debug("Added sender_name and sender_key columns with backfill")
async def _migrate_026_rename_advert_paths_table(conn: aiosqlite.Connection) -> None:
"""
Rename repeater_advert_paths to contact_advert_paths with column
repeater_key -> public_key.
Uses table rebuild since ALTER TABLE RENAME COLUMN may not be available
in older SQLite versions.
"""
# Check if old table exists
cursor = await conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='repeater_advert_paths'"
)
if not await cursor.fetchone():
# Already renamed or doesn't exist — ensure new table exists
await conn.execute(
"""
CREATE TABLE IF NOT EXISTS contact_advert_paths (
id INTEGER PRIMARY KEY AUTOINCREMENT,
public_key TEXT NOT NULL,
path_hex TEXT NOT NULL,
path_len INTEGER NOT NULL,
first_seen INTEGER NOT NULL,
last_seen INTEGER NOT NULL,
heard_count INTEGER NOT NULL DEFAULT 1,
UNIQUE(public_key, path_hex),
FOREIGN KEY (public_key) REFERENCES contacts(public_key)
)
"""
)
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_contact_advert_paths_recent "
"ON contact_advert_paths(public_key, last_seen DESC)"
)
await conn.commit()
logger.debug("contact_advert_paths already exists or old table missing, skipping rename")
return
# Create new table (IF NOT EXISTS in case SCHEMA already created it)
await conn.execute(
"""
CREATE TABLE IF NOT EXISTS contact_advert_paths (
id INTEGER PRIMARY KEY AUTOINCREMENT,
public_key TEXT NOT NULL,
path_hex TEXT NOT NULL,
path_len INTEGER NOT NULL,
first_seen INTEGER NOT NULL,
last_seen INTEGER NOT NULL,
heard_count INTEGER NOT NULL DEFAULT 1,
UNIQUE(public_key, path_hex),
FOREIGN KEY (public_key) REFERENCES contacts(public_key)
)
"""
)
# Copy data (INSERT OR IGNORE in case of duplicates)
await conn.execute(
"""
INSERT OR IGNORE INTO contact_advert_paths (public_key, path_hex, path_len, first_seen, last_seen, heard_count)
SELECT repeater_key, path_hex, path_len, first_seen, last_seen, heard_count
FROM repeater_advert_paths
"""
)
# Drop old table
await conn.execute("DROP TABLE repeater_advert_paths")
# Create index
await conn.execute(
"CREATE INDEX IF NOT EXISTS idx_contact_advert_paths_recent "
"ON contact_advert_paths(public_key, last_seen DESC)"
)
await conn.commit()
logger.info("Renamed repeater_advert_paths to contact_advert_paths")
async def _migrate_027_backfill_first_seen_from_advert_paths(conn: aiosqlite.Connection) -> None:
"""
Backfill contacts.first_seen from contact_advert_paths where advert path
first_seen is earlier than the contact's current first_seen.
"""
# Guard: skip if either table doesn't exist
for table in ("contacts", "contact_advert_paths"):
cursor = await conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table,)
)
if not await cursor.fetchone():
return
await conn.execute(
"""
UPDATE contacts SET first_seen = (
SELECT MIN(cap.first_seen) FROM contact_advert_paths cap
WHERE cap.public_key = contacts.public_key
)
WHERE EXISTS (
SELECT 1 FROM contact_advert_paths cap
WHERE cap.public_key = contacts.public_key
AND cap.first_seen < COALESCE(contacts.first_seen, 9999999999)
)
"""
)
await conn.commit()
logger.debug("Backfilled first_seen from contact_advert_paths")
app/models.py
+49 -6
View File
@@ -17,6 +17,7 @@ class Contact(BaseModel):
on_radio: bool = False
last_contacted: int | None = None # Last time we sent/received a message
last_read_at: int | None = None # Server-side read state tracking
first_seen: int | None = None
def to_radio_dict(self) -> dict:
"""Convert to the dict format expected by meshcore radio commands.
@@ -72,8 +73,8 @@ class CreateContactRequest(BaseModel):
CONTACT_TYPE_REPEATER = 2
class RepeaterAdvertPath(BaseModel):
"""A unique advert path observed for a repeater."""
class ContactAdvertPath(BaseModel):
"""A unique advert path observed for a contact."""
path: str = Field(description="Hex-encoded routing path (empty string for direct)")
path_len: int = Field(description="Number of hops in the path")
@@ -85,15 +86,57 @@ class RepeaterAdvertPath(BaseModel):
heard_count: int = Field(description="Number of times this unique path was heard")
class RepeaterAdvertPathSummary(BaseModel):
"""Recent unique advertisement paths for a single repeater."""
class ContactAdvertPathSummary(BaseModel):
"""Recent unique advertisement paths for a single contact."""
repeater_key: str = Field(description="Repeater public key (64-char hex)")
paths: list[RepeaterAdvertPath] = Field(
public_key: str = Field(description="Contact public key (64-char hex)")
paths: list[ContactAdvertPath] = Field(
default_factory=list, description="Most recent unique advert paths"
)
class ContactNameHistory(BaseModel):
"""A historical name used by a contact."""
name: str
first_seen: int
last_seen: int
class ContactActiveRoom(BaseModel):
"""A channel/room where a contact has been active."""
channel_key: str
channel_name: str
message_count: int
class NearestRepeater(BaseModel):
"""A repeater that has relayed a contact's advertisements."""
public_key: str
name: str | None = None
path_len: int
last_seen: int
heard_count: int
class ContactDetail(BaseModel):
"""Comprehensive contact profile data."""
contact: Contact
name_history: list[ContactNameHistory] = Field(default_factory=list)
dm_message_count: int = 0
channel_message_count: int = 0
most_active_rooms: list[ContactActiveRoom] = Field(default_factory=list)
advert_paths: list[ContactAdvertPath] = Field(default_factory=list)
advert_frequency: float | None = Field(
default=None,
description="Advert observations per hour (includes multi-path arrivals of same advert)",
)
nearest_repeaters: list[NearestRepeater] = Field(default_factory=list)
class Channel(BaseModel):
key: str = Field(description="Channel key (32-char hex)")
name: str
app/packet_processor.py
+26 -7
View File
@@ -31,10 +31,11 @@ from app.keystore import get_private_key, get_public_key, has_private_key
from app.models import CONTACT_TYPE_REPEATER, RawPacketBroadcast, RawPacketDecryptedInfo
from app.repository import (
ChannelRepository,
ContactAdvertPathRepository,
ContactNameHistoryRepository,
ContactRepository,
MessageRepository,
RawPacketRepository,
RepeaterAdvertPathRepository,
)
from app.websocket import broadcast_error, broadcast_event
@@ -150,6 +151,13 @@ async def create_message_from_decrypted(
# Normalize channel key to uppercase for consistency
channel_key_normalized = channel_key.upper()
# Resolve sender_key: look up contact by exact name match
resolved_sender_key: str | None = None
if sender:
candidates = await ContactRepository.get_by_name(sender)
if len(candidates) == 1:
resolved_sender_key = candidates[0].public_key
# Try to create message - INSERT OR IGNORE handles duplicates atomically
msg_id = await MessageRepository.create(
msg_type="CHAN",
@@ -158,6 +166,8 @@ async def create_message_from_decrypted(
sender_timestamp=timestamp,
received_at=received,
path=path,
sender_name=sender,
sender_key=resolved_sender_key,
)
if msg_id is None:
@@ -270,6 +280,7 @@ async def create_dm_message_from_decrypted(
received_at=received,
path=path,
outgoing=outgoing,
sender_key=conversation_key if not outgoing else None,
)
if msg_id is None:
@@ -689,13 +700,20 @@ async def _process_advertisement(
advert.device_role if advert.device_role > 0 else (existing.type if existing else 0)
)
# Keep recent unique advert paths for repeaters to improve frontend identity hints.
if contact_type == CONTACT_TYPE_REPEATER:
await RepeaterAdvertPathRepository.record_observation(
repeater_key=advert.public_key.lower(),
path_hex=new_path_hex,
# Keep recent unique advert paths for all contacts.
await ContactAdvertPathRepository.record_observation(
public_key=advert.public_key.lower(),
path_hex=new_path_hex,
timestamp=timestamp,
max_paths=10,
)
# Record name history
if advert.name:
await ContactNameHistoryRepository.record_name(
public_key=advert.public_key.lower(),
name=advert.name,
timestamp=timestamp,
max_paths_per_repeater=10,
)
contact_data = {
@@ -708,6 +726,7 @@ async def _process_advertisement(
"last_seen": timestamp,
"last_path": path_hex,
"last_path_len": path_len,
"first_seen": timestamp, # COALESCE in upsert preserves existing value
}
await ContactRepository.upsert(contact_data)
app/radio_sync.py
+6
View File
@@ -251,9 +251,15 @@ async def sync_and_offload_all(mc: MeshCore) -> dict:
# Ensure default channels exist
await ensure_default_channels()
# Reload favorites and recent contacts back onto the radio immediately
# so favorited contacts don't stay in the on_radio=False limbo until the
# next advertisement arrives.
reload_result = await sync_recent_contacts_to_radio(force=True)
return {
"contacts": contacts_result,
"channels": channels_result,
"reloaded": reload_result,
}
app/repository.py
+181 -52
View File
@@ -12,11 +12,12 @@ from app.models import (
BotConfig,
Channel,
Contact,
ContactAdvertPath,
ContactAdvertPathSummary,
ContactNameHistory,
Favorite,
Message,
MessagePath,
RepeaterAdvertPath,
RepeaterAdvertPathSummary,
)
logger = logging.getLogger(__name__)
@@ -42,8 +43,9 @@ class ContactRepository:
await db.conn.execute(
"""
INSERT INTO contacts (public_key, name, type, flags, last_path, last_path_len,
last_advert, lat, lon, last_seen, on_radio, last_contacted)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
last_advert, lat, lon, last_seen, on_radio, last_contacted,
first_seen)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(public_key) DO UPDATE SET
name = COALESCE(excluded.name, contacts.name),
type = CASE WHEN excluded.type = 0 THEN contacts.type ELSE excluded.type END,
@@ -54,8 +56,9 @@ class ContactRepository:
lat = COALESCE(excluded.lat, contacts.lat),
lon = COALESCE(excluded.lon, contacts.lon),
last_seen = excluded.last_seen,
on_radio = excluded.on_radio,
last_contacted = COALESCE(excluded.last_contacted, contacts.last_contacted)
on_radio = COALESCE(excluded.on_radio, contacts.on_radio),
last_contacted = COALESCE(excluded.last_contacted, contacts.last_contacted),
first_seen = COALESCE(contacts.first_seen, excluded.first_seen)
""",
(
contact.get("public_key", "").lower(),
@@ -68,8 +71,9 @@ class ContactRepository:
contact.get("lat"),
contact.get("lon"),
contact.get("last_seen", int(time.time())),
contact.get("on_radio", False),
contact.get("on_radio"),
contact.get("last_contacted"),
contact.get("first_seen"),
),
)
await db.conn.commit()
@@ -91,6 +95,7 @@ class ContactRepository:
on_radio=bool(row["on_radio"]),
last_contacted=row["last_contacted"],
last_read_at=row["last_read_at"],
first_seen=row["first_seen"],
)
@staticmethod
@@ -148,6 +153,42 @@ class ContactRepository:
)
return None
@staticmethod
async def get_by_name(name: str) -> list[Contact]:
"""Get all contacts with the given exact name."""
cursor = await db.conn.execute("SELECT * FROM contacts WHERE name = ?", (name,))
rows = await cursor.fetchall()
return [ContactRepository._row_to_contact(row) for row in rows]
@staticmethod
async def resolve_prefixes(prefixes: list[str]) -> dict[str, Contact]:
"""Resolve multiple key prefixes to contacts in a single query.
Returns a dict mapping each prefix to its Contact, only for prefixes
that resolve uniquely (exactly one match). Ambiguous or unmatched
prefixes are omitted.
"""
if not prefixes:
return {}
normalized = [p.lower() for p in prefixes]
conditions = " OR ".join(["public_key LIKE ?"] * len(normalized))
params = [f"{p}%" for p in normalized]
cursor = await db.conn.execute(f"SELECT * FROM contacts WHERE {conditions}", params)
rows = await cursor.fetchall()
# Group by which prefix each row matches
prefix_to_rows: dict[str, list] = {p: [] for p in normalized}
for row in rows:
pk = row["public_key"]
for p in normalized:
if pk.startswith(p):
prefix_to_rows[p].append(row)
# Only include uniquely-resolved prefixes
result: dict[str, Contact] = {}
for p in normalized:
if len(prefix_to_rows[p]) == 1:
result[p] = ContactRepository._row_to_contact(prefix_to_rows[p][0])
return result
@staticmethod
async def get_all(limit: int = 100, offset: int = 0) -> list[Contact]:
cursor = await db.conn.execute(
@@ -194,10 +235,14 @@ class ContactRepository:
@staticmethod
async def delete(public_key: str) -> None:
normalized = public_key.lower()
await db.conn.execute(
"DELETE FROM contacts WHERE public_key = ?",
(public_key.lower(),),
"DELETE FROM contact_name_history WHERE public_key = ?", (normalized,)
)
await db.conn.execute(
"DELETE FROM contact_advert_paths WHERE public_key = ?", (normalized,)
)
await db.conn.execute("DELETE FROM contacts WHERE public_key = ?", (normalized,))
await db.conn.commit()
@staticmethod
@@ -241,14 +286,14 @@ class ContactRepository:
return [ContactRepository._row_to_contact(row) for row in rows]
class RepeaterAdvertPathRepository:
"""Repository for recent unique repeater advertisement paths."""
class ContactAdvertPathRepository:
"""Repository for recent unique advertisement paths per contact."""
@staticmethod
def _row_to_path(row) -> RepeaterAdvertPath:
def _row_to_path(row) -> ContactAdvertPath:
path = row["path_hex"] or ""
next_hop = path[:2].lower() if len(path) >= 2 else None
return RepeaterAdvertPath(
return ContactAdvertPath(
path=path,
path_len=row["path_len"],
next_hop=next_hop,
@@ -259,92 +304,129 @@ class RepeaterAdvertPathRepository:
@staticmethod
async def record_observation(
repeater_key: str, path_hex: str, timestamp: int, max_paths_per_repeater: int = 10
public_key: str,
path_hex: str,
timestamp: int,
max_paths: int = 10,
) -> None:
"""
Upsert a unique advert path observation for a repeater and prune to N most recent.
Upsert a unique advert path observation for a contact and prune to N most recent.
"""
if max_paths_per_repeater < 1:
max_paths_per_repeater = 1
if max_paths < 1:
max_paths = 1
normalized_key = repeater_key.lower()
normalized_key = public_key.lower()
normalized_path = path_hex.lower()
path_len = len(normalized_path) // 2
await db.conn.execute(
"""
INSERT INTO repeater_advert_paths
(repeater_key, path_hex, path_len, first_seen, last_seen, heard_count)
INSERT INTO contact_advert_paths
(public_key, path_hex, path_len, first_seen, last_seen, heard_count)
VALUES (?, ?, ?, ?, ?, 1)
ON CONFLICT(repeater_key, path_hex) DO UPDATE SET
last_seen = MAX(repeater_advert_paths.last_seen, excluded.last_seen),
ON CONFLICT(public_key, path_hex) DO UPDATE SET
last_seen = MAX(contact_advert_paths.last_seen, excluded.last_seen),
path_len = excluded.path_len,
heard_count = repeater_advert_paths.heard_count + 1
heard_count = contact_advert_paths.heard_count + 1
""",
(normalized_key, normalized_path, path_len, timestamp, timestamp),
)
# Keep only the N most recent unique paths per repeater.
# Keep only the N most recent unique paths per contact.
await db.conn.execute(
"""
DELETE FROM repeater_advert_paths
WHERE repeater_key = ?
DELETE FROM contact_advert_paths
WHERE public_key = ?
AND path_hex NOT IN (
SELECT path_hex
FROM repeater_advert_paths
WHERE repeater_key = ?
FROM contact_advert_paths
WHERE public_key = ?
ORDER BY last_seen DESC, heard_count DESC, path_len ASC, path_hex ASC
LIMIT ?
)
""",
(normalized_key, normalized_key, max_paths_per_repeater),
(normalized_key, normalized_key, max_paths),
)
await db.conn.commit()
@staticmethod
async def get_recent_for_repeater(
repeater_key: str, limit: int = 10
) -> list[RepeaterAdvertPath]:
async def get_recent_for_contact(public_key: str, limit: int = 10) -> list[ContactAdvertPath]:
cursor = await db.conn.execute(
"""
SELECT path_hex, path_len, first_seen, last_seen, heard_count
FROM repeater_advert_paths
WHERE repeater_key = ?
FROM contact_advert_paths
WHERE public_key = ?
ORDER BY last_seen DESC, heard_count DESC, path_len ASC, path_hex ASC
LIMIT ?
""",
(repeater_key.lower(), limit),
(public_key.lower(), limit),
)
rows = await cursor.fetchall()
return [RepeaterAdvertPathRepository._row_to_path(row) for row in rows]
return [ContactAdvertPathRepository._row_to_path(row) for row in rows]
@staticmethod
async def get_recent_for_all_repeaters(
limit_per_repeater: int = 10,
) -> list[RepeaterAdvertPathSummary]:
async def get_recent_for_all_contacts(
limit_per_contact: int = 10,
) -> list[ContactAdvertPathSummary]:
cursor = await db.conn.execute(
"""
SELECT repeater_key, path_hex, path_len, first_seen, last_seen, heard_count
FROM repeater_advert_paths
ORDER BY repeater_key ASC, last_seen DESC, heard_count DESC, path_len ASC, path_hex ASC
SELECT public_key, path_hex, path_len, first_seen, last_seen, heard_count
FROM contact_advert_paths
ORDER BY public_key ASC, last_seen DESC, heard_count DESC, path_len ASC, path_hex ASC
"""
)
rows = await cursor.fetchall()
grouped: dict[str, list[RepeaterAdvertPath]] = {}
grouped: dict[str, list[ContactAdvertPath]] = {}
for row in rows:
repeater_key = row["repeater_key"]
paths = grouped.get(repeater_key)
key = row["public_key"]
paths = grouped.get(key)
if paths is None:
paths = []
grouped[repeater_key] = paths
if len(paths) >= limit_per_repeater:
grouped[key] = paths
if len(paths) >= limit_per_contact:
continue
paths.append(RepeaterAdvertPathRepository._row_to_path(row))
paths.append(ContactAdvertPathRepository._row_to_path(row))
return [
RepeaterAdvertPathSummary(repeater_key=repeater_key, paths=paths)
for repeater_key, paths in grouped.items()
ContactAdvertPathSummary(public_key=key, paths=paths) for key, paths in grouped.items()
]
class ContactNameHistoryRepository:
"""Repository for contact name change history."""
@staticmethod
async def record_name(public_key: str, name: str, timestamp: int) -> None:
"""Record a name observation. Upserts: updates last_seen if name already known."""
await db.conn.execute(
"""
INSERT INTO contact_name_history (public_key, name, first_seen, last_seen)
VALUES (?, ?, ?, ?)
ON CONFLICT(public_key, name) DO UPDATE SET
last_seen = MAX(contact_name_history.last_seen, excluded.last_seen)
""",
(public_key.lower(), name, timestamp, timestamp),
)
await db.conn.commit()
@staticmethod
async def get_history(public_key: str) -> list[ContactNameHistory]:
cursor = await db.conn.execute(
"""
SELECT name, first_seen, last_seen
FROM contact_name_history
WHERE public_key = ?
ORDER BY last_seen DESC
""",
(public_key.lower(),),
)
rows = await cursor.fetchall()
return [
ContactNameHistory(
name=row["name"], first_seen=row["first_seen"], last_seen=row["last_seen"]
)
for row in rows
]
@@ -453,6 +535,8 @@ class MessageRepository:
txt_type: int = 0,
signature: str | None = None,
outgoing: bool = False,
sender_name: str | None = None,
sender_key: str | None = None,
) -> int | None:
"""Create a message, returning the ID or None if duplicate.
@@ -470,8 +554,9 @@ class MessageRepository:
cursor = await db.conn.execute(
"""
INSERT OR IGNORE INTO messages (type, conversation_key, text, sender_timestamp,
received_at, paths, txt_type, signature, outgoing)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
received_at, paths, txt_type, signature, outgoing,
sender_name, sender_key)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
msg_type,
@@ -483,6 +568,8 @@ class MessageRepository:
txt_type,
signature,
outgoing,
sender_name,
sender_key,
),
)
await db.conn.commit()
@@ -787,6 +874,48 @@ class MessageRepository:
"last_message_times": last_message_times,
}
@staticmethod
async def count_dm_messages(contact_key: str) -> int:
"""Count total DM messages for a contact."""
cursor = await db.conn.execute(
"SELECT COUNT(*) as cnt FROM messages WHERE type = 'PRIV' AND conversation_key = ?",
(contact_key.lower(),),
)
row = await cursor.fetchone()
return row["cnt"] if row else 0
@staticmethod
async def count_channel_messages_by_sender(sender_key: str) -> int:
"""Count channel messages sent by a specific contact."""
cursor = await db.conn.execute(
"SELECT COUNT(*) as cnt FROM messages WHERE type = 'CHAN' AND sender_key = ?",
(sender_key.lower(),),
)
row = await cursor.fetchone()
return row["cnt"] if row else 0
@staticmethod
async def get_most_active_rooms(sender_key: str, limit: int = 5) -> list[tuple[str, str, int]]:
"""Get channels where a contact has sent the most messages.
Returns list of (channel_key, channel_name, message_count) tuples.
"""
cursor = await db.conn.execute(
"""
SELECT m.conversation_key, COALESCE(c.name, m.conversation_key) as channel_name,
COUNT(*) as cnt
FROM messages m
LEFT JOIN channels c ON m.conversation_key = c.key
WHERE m.type = 'CHAN' AND m.sender_key = ?
GROUP BY m.conversation_key
ORDER BY cnt DESC
LIMIT ?
""",
(sender_key.lower(), limit),
)
rows = await cursor.fetchall()
return [(row["conversation_key"], row["channel_name"], row["cnt"]) for row in rows]
class RawPacketRepository:
@staticmethod
app/routers/contacts.py
+98 -17
View File
@@ -14,10 +14,13 @@ from app.models import (
CommandRequest,
CommandResponse,
Contact,
ContactActiveRoom,
ContactAdvertPath,
ContactAdvertPathSummary,
ContactDetail,
CreateContactRequest,
NearestRepeater,
NeighborInfo,
RepeaterAdvertPath,
RepeaterAdvertPathSummary,
TelemetryRequest,
TelemetryResponse,
TraceResponse,
@@ -26,9 +29,10 @@ from app.packet_processor import start_historical_dm_decryption
from app.radio import radio_manager
from app.repository import (
AmbiguousPublicKeyPrefixError,
ContactAdvertPathRepository,
ContactNameHistoryRepository,
ContactRepository,
MessageRepository,
RepeaterAdvertPathRepository,
)
from app.websocket import broadcast_error
@@ -196,13 +200,17 @@ async def list_contacts(
return await ContactRepository.get_all(limit=limit, offset=offset)
@router.get("/repeaters/advert-paths", response_model=list[RepeaterAdvertPathSummary])
@router.get("/repeaters/advert-paths", response_model=list[ContactAdvertPathSummary])
async def list_repeater_advert_paths(
limit_per_repeater: int = Query(default=10, ge=1, le=50),
) -> list[RepeaterAdvertPathSummary]:
"""List recent unique advert paths for all repeaters."""
return await RepeaterAdvertPathRepository.get_recent_for_all_repeaters(
limit_per_repeater=limit_per_repeater
) -> list[ContactAdvertPathSummary]:
"""List recent unique advert paths for all repeaters.
Note: This endpoint now returns paths for all contacts (table was renamed).
The route is kept for backward compatibility.
"""
return await ContactAdvertPathRepository.get_recent_for_all_contacts(
limit_per_contact=limit_per_repeater
)
@@ -283,25 +291,98 @@ async def create_contact(
return Contact(**contact_data)
@router.get("/{public_key}/detail", response_model=ContactDetail)
async def get_contact_detail(public_key: str) -> ContactDetail:
"""Get comprehensive contact profile data.
Returns contact info, name history, message counts, most active rooms,
advertisement paths, advert frequency, and nearest repeaters.
"""
contact = await _resolve_contact_or_404(public_key)
name_history = await ContactNameHistoryRepository.get_history(contact.public_key)
dm_count = await MessageRepository.count_dm_messages(contact.public_key)
chan_count = await MessageRepository.count_channel_messages_by_sender(contact.public_key)
active_rooms_raw = await MessageRepository.get_most_active_rooms(contact.public_key)
advert_paths = await ContactAdvertPathRepository.get_recent_for_contact(contact.public_key)
most_active_rooms = [
ContactActiveRoom(channel_key=key, channel_name=name, message_count=count)
for key, name, count in active_rooms_raw
]
# Compute advert observation rate (observations/hour) from path data.
# Note: a single advertisement can arrive via multiple paths, so this counts
# RF observations, not unique advertisement broadcasts.
advert_frequency: float | None = None
if advert_paths:
total_observations = sum(p.heard_count for p in advert_paths)
earliest = min(p.first_seen for p in advert_paths)
latest = max(p.last_seen for p in advert_paths)
span_hours = (latest - earliest) / 3600.0
if span_hours > 0:
advert_frequency = round(total_observations / span_hours, 2)
# Compute nearest repeaters from first-hop prefixes in advert paths
first_hop_stats: dict[str, dict] = {} # prefix -> {heard_count, path_len, last_seen}
for p in advert_paths:
if p.path and len(p.path) >= 2:
prefix = p.path[:2].lower()
if prefix not in first_hop_stats:
first_hop_stats[prefix] = {
"heard_count": 0,
"path_len": p.path_len,
"last_seen": p.last_seen,
}
first_hop_stats[prefix]["heard_count"] += p.heard_count
first_hop_stats[prefix]["last_seen"] = max(
first_hop_stats[prefix]["last_seen"], p.last_seen
)
# Resolve all first-hop prefixes to contacts in a single query
resolved_contacts = await ContactRepository.resolve_prefixes(list(first_hop_stats.keys()))
nearest_repeaters: list[NearestRepeater] = []
for prefix, stats in first_hop_stats.items():
resolved = resolved_contacts.get(prefix)
nearest_repeaters.append(
NearestRepeater(
public_key=resolved.public_key if resolved else prefix,
name=resolved.name if resolved else None,
path_len=stats["path_len"],
last_seen=stats["last_seen"],
heard_count=stats["heard_count"],
)
)
nearest_repeaters.sort(key=lambda r: r.heard_count, reverse=True)
return ContactDetail(
contact=contact,
name_history=name_history,
dm_message_count=dm_count,
channel_message_count=chan_count,
most_active_rooms=most_active_rooms,
advert_paths=advert_paths,
advert_frequency=advert_frequency,
nearest_repeaters=nearest_repeaters,
)
@router.get("/{public_key}", response_model=Contact)
async def get_contact(public_key: str) -> Contact:
"""Get a specific contact by public key or prefix."""
return await _resolve_contact_or_404(public_key)
@router.get("/{public_key}/advert-paths", response_model=list[RepeaterAdvertPath])
@router.get("/{public_key}/advert-paths", response_model=list[ContactAdvertPath])
async def get_contact_advert_paths(
public_key: str,
limit: int = Query(default=10, ge=1, le=50),
) -> list[RepeaterAdvertPath]:
"""List recent unique advert paths for a single repeater contact."""
) -> list[ContactAdvertPath]:
"""List recent unique advert paths for a contact."""
contact = await _resolve_contact_or_404(public_key)
if contact.type != CONTACT_TYPE_REPEATER:
raise HTTPException(
status_code=400,
detail=f"Contact is not a repeater (type={contact.type}, expected {CONTACT_TYPE_REPEATER})",
)
return await RepeaterAdvertPathRepository.get_recent_for_repeater(contact.public_key, limit)
return await ContactAdvertPathRepository.get_recent_for_contact(contact.public_key, limit)
@router.post("/sync")
app/routers/settings.py
+12 -2
View File
@@ -1,3 +1,4 @@
import asyncio
import logging
from typing import Literal
@@ -137,10 +138,19 @@ async def toggle_favorite(request: FavoriteRequest) -> AppSettings:
if is_favorited:
logger.info("Removing favorite: %s %s", request.type, request.id[:12])
return await AppSettingsRepository.remove_favorite(request.type, request.id)
result = await AppSettingsRepository.remove_favorite(request.type, request.id)
else:
logger.info("Adding favorite: %s %s", request.type, request.id[:12])
return await AppSettingsRepository.add_favorite(request.type, request.id)
result = await AppSettingsRepository.add_favorite(request.type, request.id)
# When a contact favorite changes, sync the radio so the contact is
# loaded/unloaded immediately rather than waiting for the next advert.
if request.type == "contact":
from app.radio_sync import sync_recent_contacts_to_radio
asyncio.create_task(sync_recent_contacts_to_radio(force=True))
return result
@router.post("/migrate", response_model=MigratePreferencesResponse)