hardening and cleanup

Author: talvasconcelos

__init__.py

@@ -9,9 +9,7 @@
 from .views import market_town_generic_router
 from .views_api import market_town_api_router
 
-market_town_ext: APIRouter = APIRouter(
-    prefix="/market_town", tags=["Market Town"]
-)
+market_town_ext: APIRouter = APIRouter(prefix="/market_town", tags=["Market Town"])
 market_town_ext.include_router(market_town_generic_router)
 market_town_ext.include_router(market_town_api_router)
 
@@ -35,12 +33,8 @@ def market_town_stop():
 
 
 def market_town_start():
-    scheduled_tasks.append(
-        create_permanent_unique_task("ext_market_town_paid_invoices", wait_for_paid_invoices)
-    )
-    scheduled_tasks.append(
-        create_permanent_unique_task("ext_market_town_scheduler", run_scheduler_loop)
-    )
+    scheduled_tasks.append(create_permanent_unique_task("ext_market_town_paid_invoices", wait_for_paid_invoices))
+    scheduled_tasks.append(create_permanent_unique_task("ext_market_town_scheduler", run_scheduler_loop))
 
 
 __all__ = [

crud.py

@@ -260,6 +260,10 @@ async def create_agent(
     return agent
 
 
+async def delete_agent(agent_id: str) -> None:
+    await db.execute("DELETE FROM market_town.agents WHERE id = :agent_id", {"agent_id": agent_id})
+
+
 async def get_agent(agent_id: str) -> Agent | None:
     return await db.fetchone(
         "SELECT * FROM market_town.agents WHERE id = :id",
@@ -318,6 +322,13 @@ async def create_business(
     return business
 
 
+async def delete_business(business_id: str) -> None:
+    await db.execute(
+        "DELETE FROM market_town.businesses WHERE id = :business_id",
+        {"business_id": business_id},
+    )
+
+
 async def get_business(business_id: str) -> Business | None:
     return await db.fetchone(
         "SELECT * FROM market_town.businesses WHERE id = :id",
@@ -381,6 +392,33 @@ async def get_epoch(world_id: str, epoch_number: int) -> Epoch | None:
     )
 
 
+async def claim_epoch_for_resolution(world_id: str, epoch_number: int, stale_before: datetime) -> bool:
+    result = await db.execute(
+        f"""
+        UPDATE market_town.epochs
+        SET status = 'resolving',
+            updated_at = {db.timestamp_placeholder("now")}
+        WHERE world_id = :world_id
+          AND epoch_number = :epoch_number
+          AND resolved_at IS NULL
+          AND (
+            status = 'open'
+            OR (
+              status = 'resolving'
+              AND updated_at <= {db.timestamp_placeholder("stale_before")}
+            )
+          )
+        """,
+        {
+            "world_id": world_id,
+            "epoch_number": epoch_number,
+            "now": utc_now().timestamp(),
+            "stale_before": stale_before.timestamp(),
+        },
+    )
+    return bool(result.rowcount)
+
+
 async def get_latest_epoch(world_id: str) -> Epoch | None:
     return await db.fetchone(
         """
@@ -511,6 +549,26 @@ async def create_snapshot(snapshot: BusinessEpochSnapshot) -> BusinessEpochSnaps
     return snapshot
 
 
+async def get_snapshot_for_business_epoch(
+    world_id: str, epoch_number: int, business_id: str
+) -> BusinessEpochSnapshot | None:
+    return await db.fetchone(
+        """
+        SELECT * FROM market_town.business_epoch_snapshots
+        WHERE world_id = :world_id
+          AND epoch_number = :epoch_number
+          AND business_id = :business_id
+        LIMIT 1
+        """,
+        {
+            "world_id": world_id,
+            "epoch_number": epoch_number,
+            "business_id": business_id,
+        },
+        BusinessEpochSnapshot,
+    )
+
+
 async def list_snapshots_for_business(business_id: str, limit: int = 10) -> list[BusinessEpochSnapshot]:
     return await db.fetchall(
         f"""
@@ -528,6 +586,18 @@ async def create_season_result(season_result: SeasonResult) -> SeasonResult:
     return season_result
 
 
+async def get_season_result(world_id: str, season_number: int) -> SeasonResult | None:
+    return await db.fetchone(
+        """
+        SELECT * FROM market_town.season_results
+        WHERE world_id = :world_id AND season_number = :season_number
+        LIMIT 1
+        """,
+        {"world_id": world_id, "season_number": season_number},
+        SeasonResult,
+    )
+
+
 async def update_season_result(season_result: SeasonResult) -> SeasonResult:
     updated = season_result.copy(update={"updated_at": utc_now()})
     await db.update("market_town.season_results", updated)
@@ -553,11 +623,7 @@ async def list_paid_payment_requests_for_season(
     *,
     include_before_start: bool = False,
 ) -> list[PaymentRequestRecord]:
-    lower_clause = (
-        ""
-        if include_before_start
-        else f"AND paid_at >= {db.timestamp_placeholder('season_started_at')}"
-    )
+    lower_clause = "" if include_before_start else f"AND paid_at >= {db.timestamp_placeholder('season_started_at')}"
     return await db.fetchall(
         f"""
         SELECT * FROM market_town.payment_requests
@@ -602,6 +668,35 @@ async def get_payment_request_by_hash(payment_hash: str) -> PaymentRequestRecord
     )
 
 
+async def claim_payment_request_for_settlement(
+    payment_hash: str, stale_before: datetime
+) -> PaymentRequestRecord | None:
+    result = await db.execute(
+        f"""
+        UPDATE market_town.payment_requests
+        SET status = 'settling',
+            updated_at = {db.timestamp_placeholder("now")}
+        WHERE payment_hash = :payment_hash
+          AND (
+            status = 'pending'
+            OR status = 'expired'
+            OR (
+              status = 'settling'
+              AND updated_at <= {db.timestamp_placeholder("stale_before")}
+            )
+          )
+        """,
+        {
+            "payment_hash": payment_hash,
+            "now": utc_now().timestamp(),
+            "stale_before": stale_before.timestamp(),
+        },
+    )
+    if not result.rowcount:
+        return None
+    return await get_payment_request_by_hash(payment_hash)
+
+
 async def get_payment_request_by_claim_token(claim_token: str) -> PaymentRequestRecord | None:
     return await db.fetchone(
         """
@@ -614,6 +709,41 @@ async def get_payment_request_by_claim_token(claim_token: str) -> PaymentRequest
     )
 
 
+async def claim_payment_request_credentials_reveal(
+    claim_token: str,
+) -> PaymentRequestRecord | None:
+    result = await db.execute(
+        f"""
+        UPDATE market_town.payment_requests
+        SET credentials_revealed = TRUE,
+            issued_api_key = NULL,
+            updated_at = {db.timestamp_placeholder("now")}
+        WHERE claim_token = :claim_token
+          AND status = 'paid'
+          AND credentials_revealed = FALSE
+          AND agent_id IS NOT NULL
+          AND business_id IS NOT NULL
+        """,
+        {"claim_token": claim_token, "now": utc_now().timestamp()},
+    )
+    if not result.rowcount:
+        return None
+    return await get_payment_request_by_claim_token(claim_token)
+
+
+async def reset_payment_request_credentials_reveal(claim_token: str) -> None:
+    await db.execute(
+        f"""
+        UPDATE market_town.payment_requests
+        SET credentials_revealed = FALSE,
+            updated_at = {db.timestamp_placeholder("now")}
+        WHERE claim_token = :claim_token
+          AND credentials_revealed = TRUE
+        """,
+        {"claim_token": claim_token, "now": utc_now().timestamp()},
+    )
+
+
 async def list_pending_payment_requests(world_id: str) -> list[PaymentRequestRecord]:
     return await db.fetchall(
         """
@@ -626,6 +756,53 @@ async def list_pending_payment_requests(world_id: str) -> list[PaymentRequestRec
     )
 
 
+async def list_active_pending_payment_requests(world_id: str, before_time: datetime) -> list[PaymentRequestRecord]:
+    now = utc_now()
+    return await db.fetchall(
+        f"""
+        SELECT * FROM market_town.payment_requests
+        WHERE world_id = :world_id
+          AND status = 'pending'
+          AND created_at > {db.timestamp_placeholder("before_time")}
+          AND (
+            reservation_expires_at IS NULL
+            OR reservation_expires_at > {db.timestamp_placeholder("now")}
+          )
+        ORDER BY created_at DESC
+        """,
+        {
+            "world_id": world_id,
+            "before_time": before_time.timestamp(),
+            "now": now.timestamp(),
+        },
+        PaymentRequestRecord,
+    )
+
+
+async def expire_pending_payment_requests(world_id: str, before_time: datetime) -> None:
+    await db.execute(
+        f"""
+        UPDATE market_town.payment_requests
+        SET status = 'expired',
+            updated_at = {db.timestamp_placeholder("now")}
+        WHERE world_id = :world_id
+          AND status = 'pending'
+          AND (
+            created_at <= {db.timestamp_placeholder("before_time")}
+            OR (
+              reservation_expires_at IS NOT NULL
+              AND reservation_expires_at <= {db.timestamp_placeholder("now")}
+            )
+          )
+        """,
+        {
+            "world_id": world_id,
+            "now": utc_now().timestamp(),
+            "before_time": before_time.timestamp(),
+        },
+    )
+
+
 async def update_payment_request(payment_request: PaymentRequestRecord) -> PaymentRequestRecord:
     updated = payment_request.copy(update={"updated_at": utc_now()})
     await db.update("market_town.payment_requests", updated)

market-town-player/docs/storage-and-secrets.md

@@ -80,7 +80,14 @@ Logs should be JSON Lines files.
 `logs/decisions.jsonl` may contain:
 
 ```json
-{"epoch":1,"price_sat":220,"restock_units":40,"maintenance_budget_sat":6,"quality_budget_sat":5,"reason":"initial conservative policy"}
+{
+  "epoch": 1,
+  "price_sat": 220,
+  "restock_units": 40,
+  "maintenance_budget_sat": 6,
+  "quality_budget_sat": 5,
+  "reason": "initial conservative policy"
+}
 ```
 
 `logs/errors.jsonl` may contain non-secret error details.

migrations.py

@@ -26,7 +26,8 @@ async def m001_initial(db):
             last_digest_text TEXT,
             started_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
             created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
-            updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
+            updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
+            UNIQUE (user_id)
         );
         """
     )
@@ -45,7 +46,8 @@ async def m001_initial(db):
             slot_limit INTEGER NOT NULL DEFAULT 10,
             config_text TEXT,
             created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
-            updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
+            updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
+            UNIQUE (world_id, district_key)
         );
         """
     )
@@ -64,7 +66,8 @@ async def m001_initial(db):
             base_capacity_units INTEGER NOT NULL DEFAULT 20,
             config_text TEXT,
             created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
-            updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
+            updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
+            UNIQUE (world_id, type_key)
         );
         """
     )
@@ -81,7 +84,8 @@ async def m001_initial(db):
             last_claimed_at TIMESTAMP,
             last_opened_at TIMESTAMP,
             created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
-            updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
+            updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
+            UNIQUE (world_id, api_key_hash)
         );
         """
     )
@@ -126,7 +130,8 @@ async def m001_initial(db):
             event_summary_text TEXT,
             digest_text TEXT,
             created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
-            updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
+            updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
+            UNIQUE (world_id, epoch_number)
         );
         """
     )
@@ -166,7 +171,8 @@ async def m001_initial(db):
             reliability_after REAL NOT NULL DEFAULT 0,
             quality_before REAL NOT NULL DEFAULT 0,
             quality_after REAL NOT NULL DEFAULT 0,
-            created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
+            created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
+            UNIQUE (world_id, epoch_number, business_id)
         );
         """
     )
@@ -183,7 +189,8 @@ async def m001_initial(db):
             payout_status TEXT NOT NULL DEFAULT 'pending',
             payout_summary_text TEXT,
             created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
-            updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
+            updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
+            UNIQUE (world_id, season_number)
         );
         """
     )
@@ -211,7 +218,9 @@ async def m001_initial(db):
             credentials_revealed BOOLEAN NOT NULL DEFAULT FALSE,
             paid_at TIMESTAMP,
             created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
-            updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
+            updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
+            UNIQUE (payment_hash),
+            UNIQUE (claim_token)
         );
         """
     )
@@ -228,3 +237,7 @@ async def m001_initial(db):
         );
         """
     )
+
+
+async def m002_add_payment_request_reservation_expiry(db):
+    await db.execute("ALTER TABLE market_town.payment_requests ADD COLUMN reservation_expires_at TIMESTAMP;")