Add server-side read management

app/CLAUDE.md

@@ -17,7 +17,8 @@ This document provides context for AI assistants and developers working on the F
 app/
 ├── main.py           # FastAPI app, lifespan, router registration, static file serving
 ├── config.py         # Pydantic settings (env vars: MESHCORE_*)
-├── database.py       # SQLite schema, connection management
+├── database.py       # SQLite schema, connection management, runs migrations
+├── migrations.py     # Database migrations using SQLite user_version pragma
 ├── models.py         # Pydantic models for API request/response
 ├── repository.py     # Database CRUD (ContactRepository, ChannelRepository, etc.)
 ├── radio.py          # RadioManager - serial connection to MeshCore device
@@ -30,10 +31,11 @@ app/
 └── routers/          # All routes prefixed with /api
     ├── health.py     # GET /api/health
     ├── radio.py      # Radio config, advertise, private key, reboot
-    ├── contacts.py   # Contact CRUD and radio sync
-    ├── channels.py   # Channel CRUD and radio sync
+    ├── contacts.py   # Contact CRUD, radio sync, mark-read
+    ├── channels.py   # Channel CRUD, radio sync, mark-read
     ├── messages.py   # Message list and send (direct/channel)
     ├── packets.py    # Raw packet endpoints, historical decryption
+    ├── read_state.py # Bulk read state operations (mark-all-read)
     ├── settings.py   # App settings (max_radio_contacts)
     └── ws.py         # WebSocket endpoint at /api/ws
 ```
@@ -138,14 +140,16 @@ contacts (
     lat REAL, lon REAL,
     last_seen INTEGER,
     on_radio INTEGER DEFAULT 0,   -- Boolean: contact loaded on radio
-    last_contacted INTEGER        -- Unix timestamp of last message sent/received
+    last_contacted INTEGER,       -- Unix timestamp of last message sent/received
+    last_read_at INTEGER          -- Unix timestamp when conversation was last read
 )
 
 channels (
     key TEXT PRIMARY KEY,         -- 32-char hex channel key
     name TEXT NOT NULL,
     is_hashtag INTEGER DEFAULT 0, -- Key derived from SHA256(name)[:16]
-    on_radio INTEGER DEFAULT 0
+    on_radio INTEGER DEFAULT 0,
+    last_read_at INTEGER          -- Unix timestamp when channel was last read
 )
 
 messages (
@@ -175,6 +179,49 @@ raw_packets (
 )
 ```
 
+## Database Migrations (`migrations.py`)
+
+Schema migrations use SQLite's `user_version` pragma for version tracking:
+
+```python
+from app.migrations import get_version, set_version, run_migrations
+
+# Check current schema version
+version = await get_version(conn)  # Returns int (0 for new/unmigrated DB)
+
+# Run pending migrations (called automatically on startup)
+applied = await run_migrations(conn)  # Returns number of migrations applied
+```
+
+### How It Works
+
+1. `database.py` calls `run_migrations()` after schema initialization
+2. Each migration checks `user_version` and runs if needed
+3. Migrations are idempotent (safe to run multiple times)
+4. `ALTER TABLE ADD COLUMN` handles existing columns gracefully
+
+### Adding a New Migration
+
+```python
+# In migrations.py
+async def run_migrations(conn: aiosqlite.Connection) -> int:
+    version = await get_version(conn)
+    applied = 0
+
+    if version < 1:
+        await _migrate_001_add_last_read_at(conn)
+        await set_version(conn, 1)
+        applied += 1
+
+    # Add new migrations here:
+    # if version < 2:
+    #     await _migrate_002_something(conn)
+    #     await set_version(conn, 2)
+    #     applied += 1
+
+    return applied
+```
+
 ## Packet Decryption (`decoder.py`)
 
 The decoder handles MeshCore packet decryption for historical packet analysis:
@@ -326,6 +373,7 @@ All endpoints are prefixed with `/api`.
 - `POST /api/contacts/sync` - Pull from radio to database
 - `POST /api/contacts/{key}/add-to-radio` - Push to radio
 - `POST /api/contacts/{key}/remove-from-radio` - Remove from radio
+- `POST /api/contacts/{key}/mark-read` - Mark conversation as read (updates last_read_at)
 - `POST /api/contacts/{key}/telemetry` - Request telemetry from repeater (see below)
 
 ### Channels
@@ -333,8 +381,12 @@ All endpoints are prefixed with `/api`.
 - `GET /api/channels/{key}` - Get by channel key
 - `POST /api/channels` - Create (hashtag if name starts with # or no key provided)
 - `POST /api/channels/sync` - Pull from radio
+- `POST /api/channels/{key}/mark-read` - Mark channel as read (updates last_read_at)
 - `DELETE /api/channels/{key}` - Delete channel
 
+### Read State
+- `POST /api/read-state/mark-all-read` - Mark all contacts and channels as read
+
 ### Messages
 - `GET /api/messages?type=&conversation_key=&limit=&offset=` - List with filters
 - `POST /api/messages/direct` - Send direct message
@@ -368,7 +420,8 @@ Key test files:
 - `tests/test_decoder.py` - Channel + direct message decryption, key exchange, real-world test vectors
 - `tests/test_keystore.py` - Ephemeral key store operations
 - `tests/test_event_handlers.py` - ACK tracking, repeat detection, CLI response filtering
-- `tests/test_api.py` - API endpoint tests
+- `tests/test_api.py` - API endpoint tests, read state tracking
+- `tests/test_migrations.py` - Migration system, schema versioning
 
 ## Common Tasks
 

app/database.py

@@ -77,6 +77,10 @@ class Database:
         await self._connection.commit()
         logger.debug("Database schema initialized")
 
+        # Run any pending migrations
+        from app.migrations import run_migrations
+        await run_migrations(self._connection)
+
     async def disconnect(self) -> None:
         if self._connection:
             await self._connection.close()

app/main.py

@@ -19,7 +19,7 @@ from app.radio_sync import (
     stop_periodic_sync,
     sync_and_offload_all,
 )
-from app.routers import channels, contacts, health, messages, packets, radio, settings, ws
+from app.routers import channels, contacts, health, messages, packets, radio, read_state, settings, ws
 
 setup_logging()
 logger = logging.getLogger(__name__)
@@ -100,6 +100,7 @@ app.include_router(contacts.router, prefix="/api")
 app.include_router(channels.router, prefix="/api")
 app.include_router(messages.router, prefix="/api")
 app.include_router(packets.router, prefix="/api")
+app.include_router(read_state.router, prefix="/api")
 app.include_router(settings.router, prefix="/api")
 app.include_router(ws.router, prefix="/api")
 

app/migrations.py

@@ -0,0 +1,90 @@
+"""
+Database migrations using SQLite's user_version pragma.
+
+Migrations run automatically on startup. The user_version pragma tracks
+which migrations have been applied (defaults to 0 for existing databases).
+
+This approach is safe for existing users - their databases have user_version=0,
+so all migrations run in order on first startup after upgrade.
+"""
+
+import logging
+
+import aiosqlite
+
+logger = logging.getLogger(__name__)
+
+
+async def get_version(conn: aiosqlite.Connection) -> int:
+    """Get current schema version from SQLite user_version pragma."""
+    cursor = await conn.execute("PRAGMA user_version")
+    row = await cursor.fetchone()
+    return row[0] if row else 0
+
+
+async def set_version(conn: aiosqlite.Connection, version: int) -> None:
+    """Set schema version using SQLite user_version pragma."""
+    await conn.execute(f"PRAGMA user_version = {version}")
+
+
+async def run_migrations(conn: aiosqlite.Connection) -> int:
+    """
+    Run all pending migrations.
+
+    Returns the number of migrations applied.
+    """
+    version = await get_version(conn)
+    applied = 0
+
+    # Migration 1: Add last_read_at columns for server-side read tracking
+    if version < 1:
+        logger.info("Applying migration 1: add last_read_at columns")
+        await _migrate_001_add_last_read_at(conn)
+        await set_version(conn, 1)
+        applied += 1
+
+    # Future migrations go here:
+    # if version < 2:
+    #     await _migrate_002_something(conn)
+    #     await set_version(conn, 2)
+    #     applied += 1
+
+    if applied > 0:
+        logger.info("Applied %d migration(s), schema now at version %d", applied, await get_version(conn))
+    else:
+        logger.debug("Schema up to date at version %d", version)
+
+    return applied
+
+
+async def _migrate_001_add_last_read_at(conn: aiosqlite.Connection) -> None:
+    """
+    Add last_read_at column to contacts and channels tables.
+
+    This enables server-side read state tracking, replacing the localStorage
+    approach for consistent read state across devices.
+
+    ALTER TABLE ADD COLUMN is safe - it preserves existing data and handles
+    the "column already exists" case gracefully.
+    """
+    # Add to contacts table
+    try:
+        await conn.execute("ALTER TABLE contacts ADD COLUMN last_read_at INTEGER")
+        logger.debug("Added last_read_at to contacts table")
+    except aiosqlite.OperationalError as e:
+        if "duplicate column name" in str(e).lower():
+            logger.debug("contacts.last_read_at already exists, skipping")
+        else:
+            raise
+
+    # Add to channels table
+    try:
+        await conn.execute("ALTER TABLE channels ADD COLUMN last_read_at INTEGER")
+        logger.debug("Added last_read_at to channels table")
+    except aiosqlite.OperationalError as e:
+        if "duplicate column name" in str(e).lower():
+            logger.debug("channels.last_read_at already exists, skipping")
+        else:
+            raise
+
+    await conn.commit()

app/models.py

@@ -14,6 +14,7 @@ class Contact(BaseModel):
     last_seen: int | None = None
     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
 
     def to_radio_dict(self) -> dict:
         """Convert to the dict format expected by meshcore radio commands.
@@ -63,6 +64,7 @@ class Channel(BaseModel):
     name: str
     is_hashtag: bool = False
     on_radio: bool = False
+    last_read_at: int | None = None  # Server-side read state tracking
 
 
 class Message(BaseModel):

app/repository.py

@@ -59,6 +59,7 @@ class ContactRepository:
             last_seen=row["last_seen"],
             on_radio=bool(row["on_radio"]),
             last_contacted=row["last_contacted"],
+            last_read_at=row["last_read_at"],
         )
 
     @staticmethod
@@ -169,6 +170,20 @@ class ContactRepository:
         )
         await db.conn.commit()
 
+    @staticmethod
+    async def update_last_read_at(public_key: str, timestamp: int | None = None) -> bool:
+        """Update the last_read_at timestamp for a contact.
+
+        Returns True if a row was updated, False if contact not found.
+        """
+        ts = timestamp or int(time.time())
+        cursor = await db.conn.execute(
+            "UPDATE contacts SET last_read_at = ? WHERE public_key = ?",
+            (ts, public_key),
+        )
+        await db.conn.commit()
+        return cursor.rowcount > 0
+
 
 class ChannelRepository:
     @staticmethod
@@ -191,7 +206,7 @@ class ChannelRepository:
     async def get_by_key(key: str) -> Channel | None:
         """Get a channel by its key (32-char hex string)."""
         cursor = await db.conn.execute(
-            "SELECT key, name, is_hashtag, on_radio FROM channels WHERE key = ?",
+            "SELECT key, name, is_hashtag, on_radio, last_read_at FROM channels WHERE key = ?",
             (key.upper(),)
         )
         row = await cursor.fetchone()
@@ -201,13 +216,14 @@ class ChannelRepository:
                 name=row["name"],
                 is_hashtag=bool(row["is_hashtag"]),
                 on_radio=bool(row["on_radio"]),
+                last_read_at=row["last_read_at"],
             )
         return None
 
     @staticmethod
     async def get_all() -> list[Channel]:
         cursor = await db.conn.execute(
-            "SELECT key, name, is_hashtag, on_radio FROM channels ORDER BY name"
+            "SELECT key, name, is_hashtag, on_radio, last_read_at FROM channels ORDER BY name"
         )
         rows = await cursor.fetchall()
         return [
@@ -216,6 +232,7 @@ class ChannelRepository:
                 name=row["name"],
                 is_hashtag=bool(row["is_hashtag"]),
                 on_radio=bool(row["on_radio"]),
+                last_read_at=row["last_read_at"],
             )
             for row in rows
         ]
@@ -224,7 +241,7 @@ class ChannelRepository:
     async def get_by_name(name: str) -> Channel | None:
         """Get a channel by name."""
         cursor = await db.conn.execute(
-            "SELECT key, name, is_hashtag, on_radio FROM channels WHERE name = ?", (name,)
+            "SELECT key, name, is_hashtag, on_radio, last_read_at FROM channels WHERE name = ?", (name,)
         )
         row = await cursor.fetchone()
         if row:
@@ -233,6 +250,7 @@ class ChannelRepository:
                 name=row["name"],
                 is_hashtag=bool(row["is_hashtag"]),
                 on_radio=bool(row["on_radio"]),
+                last_read_at=row["last_read_at"],
             )
         return None
 
@@ -245,6 +263,20 @@ class ChannelRepository:
         )
         await db.conn.commit()
 
+    @staticmethod
+    async def update_last_read_at(key: str, timestamp: int | None = None) -> bool:
+        """Update the last_read_at timestamp for a channel.
+
+        Returns True if a row was updated, False if channel not found.
+        """
+        ts = timestamp or int(time.time())
+        cursor = await db.conn.execute(
+            "UPDATE channels SET last_read_at = ? WHERE key = ?",
+            (ts, key.upper()),
+        )
+        await db.conn.commit()
+        return cursor.rowcount > 0
+
 
 class MessageRepository:
     @staticmethod

app/routers/channels.py

@@ -118,6 +118,20 @@ async def sync_channels_from_radio(
     return {"synced": count}
 
 
+@router.post("/{key}/mark-read")
+async def mark_channel_read(key: str) -> dict:
+    """Mark a channel as read (update last_read_at timestamp)."""
+    channel = await ChannelRepository.get_by_key(key)
+    if not channel:
+        raise HTTPException(status_code=404, detail="Channel not found")
+
+    updated = await ChannelRepository.update_last_read_at(key)
+    if not updated:
+        raise HTTPException(status_code=500, detail="Failed to update read state")
+
+    return {"status": "ok", "key": channel.key}
+
+
 @router.delete("/{key}")
 async def delete_channel(key: str) -> dict:
     """Delete a channel from the database by key.

app/routers/contacts.py

@@ -214,6 +214,20 @@ async def add_contact_to_radio(public_key: str) -> dict:
     return {"status": "ok"}
 
 
+@router.post("/{public_key}/mark-read")
+async def mark_contact_read(public_key: str) -> dict:
+    """Mark a contact conversation as read (update last_read_at timestamp)."""
+    contact = await ContactRepository.get_by_key_or_prefix(public_key)
+    if not contact:
+        raise HTTPException(status_code=404, detail="Contact not found")
+
+    updated = await ContactRepository.update_last_read_at(contact.public_key)
+    if not updated:
+        raise HTTPException(status_code=500, detail="Failed to update read state")
+
+    return {"status": "ok", "public_key": contact.public_key}
+
+
 @router.delete("/{public_key}")
 async def delete_contact(public_key: str) -> dict:
     """Delete a contact from the database (and radio if present)."""

app/routers/read_state.py

@@ -0,0 +1,35 @@
+"""Read state management endpoints."""
+
+import logging
+import time
+
+from fastapi import APIRouter
+
+from app.database import db
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/read-state", tags=["read-state"])
+
+
+@router.post("/mark-all-read")
+async def mark_all_read() -> dict:
+    """Mark all contacts and channels as read.
+
+    Updates last_read_at to current timestamp for all contacts and channels
+    in a single database transaction.
+    """
+    now = int(time.time())
+
+    # Update all contacts and channels in one transaction
+    await db.conn.execute(
+        "UPDATE contacts SET last_read_at = ?",
+        (now,)
+    )
+    await db.conn.execute(
+        "UPDATE channels SET last_read_at = ?",
+        (now,)
+    )
+    await db.conn.commit()
+
+    logger.info("Marked all contacts and channels as read at %d", now)
+    return {"status": "ok", "timestamp": now}