Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343"""Tuner web app: live sliders + canvas to iterate on sim rules.
Run: python server.pyThen open http://localhost:8920 in a browser."""from __future__ import annotations
import argparseimport functoolsimport osimport reimport sqlite3import subprocessimport threadingimport timefrom collections import defaultdict, dequefrom pathlib import Path
import json as _stdjsonfrom flask import Flask, Response, jsonify, request, send_from_directory
def fast_json(payload, status=200): """jsonify is slow for big payloads — use stdlib json + a Response.""" return Response(_stdjson.dumps(payload, separators=(",", ":")), status=status, mimetype="application/json")
from sim import rulesfrom sim.crops import CROP_NAMESfrom sim.world import SCATTER_PARAMS
# ---- input validation ----
# bsky handles are domain-like: letters/digits/dots/dashes, 2-253 charsHANDLE_RE = re.compile(r"^[a-z0-9][a-z0-9.-]{0,252}$", re.IGNORECASE)DID_RE = re.compile(r"^did:[a-z]+:[A-Za-z0-9._:-]+$")
def valid_handle(h): return isinstance(h, str) and bool(HANDLE_RE.match(h))
def valid_did(d): return isinstance(d, str) and bool(DID_RE.match(d))
def clamp_int(v, lo, hi, default): try: n = int(v) except (TypeError, ValueError): return default return max(lo, min(hi, n))
def clean_did_list(dids, cap=500): if not isinstance(dids, list): return None out = [d for d in dids if valid_did(d)] return out[:cap] if out else []
# ---- rate limiter (sliding 60s window, per-IP per-endpoint) ----
_RL_BUCKETS: dict[tuple, deque] = defaultdict(deque)_RL_LOCK = threading.Lock()
def rate_limit(per_minute: int): def decorator(view): @functools.wraps(view) def wrapped(*args, **kwargs): xff = request.headers.get("X-Forwarded-For", "") or "" ip = xff.split(",")[0].strip() or (request.remote_addr or "0") now = time.time() cutoff = now - 60 key = (view.__name__, ip) with _RL_LOCK: bucket = _RL_BUCKETS[key] while bucket and bucket[0] < cutoff: bucket.popleft() if len(bucket) >= per_minute: return ("rate limit exceeded", 429) bucket.append(now) return view(*args, **kwargs) return wrapped return decorator
HERE = Path(__file__).parentSTATIC = HERE / "static"DEFAULT_DB = HERE / "data" / "farm.db"
def _build_id() -> str: """Short git SHA + dirty marker, captured at startup.
Prefers a .build_id file dropped by tools/push at packaging time, since the Docker image has no git binary or .git directory. """ stamp = STATIC / ".build_id" if not stamp.is_file(): stamp = HERE / ".build_id" if stamp.is_file(): try: v = stamp.read_text().strip() if v: return v except Exception: pass try: return subprocess.check_output( ["git", "rev-parse", "--short", "HEAD"], cwd=HERE, stderr=subprocess.DEVNULL, ).decode().strip() except Exception: return "unknown"
BUILD_ID = _build_id()STARTED_AT = time.time()
# ---- per-thread WAL connection pool ----
_DB_LOCAL = threading.local()
def get_db(db_path: Path) -> sqlite3.Connection: """Return a thread-local SQLite connection to db_path.
WAL + relaxed sync + 1 GB mmap. The fetcher writes concurrently with these read paths, so WAL is what keeps readers from blocking. """ conn = getattr(_DB_LOCAL, "conn", None) if conn is None or getattr(_DB_LOCAL, "path", None) != str(db_path): if conn is not None: conn.close() conn = sqlite3.connect(str(db_path)) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA synchronous=NORMAL") conn.execute("PRAGMA mmap_size=1073741824") # ensure composite (author_did, created_at) index exists for the # bulk event load — adds at most a few seconds on first start. conn.execute("CREATE INDEX IF NOT EXISTS idx_posts_author_created ON posts(author_did, created_at)") conn.commit() _DB_LOCAL.conn = conn _DB_LOCAL.path = str(db_path) return conn
def create_app(db_path: Path) -> Flask: app = Flask(__name__, static_folder=str(STATIC), static_url_path="/static") # cap request bodies at 256 KB — no /api/* endpoint needs more than that app.config["MAX_CONTENT_LENGTH"] = 256_000
@app.get("/") def index(): return send_from_directory(STATIC, "index.html")
@app.get("/api/config") @rate_limit(120) def get_config(): """Return default tunables + the list of known crop names.""" return jsonify({ "defaults": rules.get_config(), "tunable_names": rules.TUNABLES, "crops": CROP_NAMES, "home_radius_default": 0, "scatter_default": SCATTER_PARAMS, "top_neighbors_default": 12, "build_id": BUILD_ID, "started_at": STARTED_AT, })
@app.get("/api/handles") @rate_limit(120) def get_handles(): """Prefix-search handles in the cache. Powers the focus-user typeahead.""" q = (request.args.get("q") or "").strip().lower() limit = clamp_int(request.args.get("limit", 12), 1, 50, 12) if not q or len(q) < 1: return jsonify({"handles": []}) # only handles that actually have own posts in the cache — otherwise the # focus lookup would fail anyway if not db_path.exists(): return jsonify({"handles": []}) db = get_db(db_path) rows = db.execute( """SELECT p.handle, COUNT(po.uri) AS post_count FROM profiles p LEFT JOIN posts po ON po.author_did = p.did WHERE p.did <> '' AND LOWER(p.handle) LIKE ? || '%' GROUP BY p.did HAVING post_count > 0 ORDER BY post_count DESC LIMIT ?""", (q, limit), ).fetchall() return jsonify({"handles": [r["handle"] for r in rows]})
@app.post("/api/neighbors") @rate_limit(60) def post_neighbors(): """Ranked candidate neighbors for a focus user — feeds the picker UI.""" body = request.get_json(silent=True) or {} handle = body.get("handle") or "cee.wtf" if not valid_handle(handle): return jsonify({"focus_did": None, "neighbors": []}), 400 limit = clamp_int(body.get("limit", 200), 1, 500, 200)
if not db_path.exists(): return jsonify({"focus_did": None, "neighbors": []}) db = get_db(db_path) row = db.execute("SELECT did FROM profiles WHERE handle = ? AND did <> ''", (handle,)).fetchone() if not row or not row["did"]: return jsonify({"focus_did": None, "neighbors": []}) focus_did = row["did"] from sim.world import neighbor_query neighbors = neighbor_query(db, focus_did, limit) return jsonify({ "focus_did": focus_did, "neighbors": [ { "did": n.did, "handle": n.handle, "social_class": n.social_class, "engagement_count": n.engagement_count, } for n in neighbors ], })
@app.get("/api/users") def get_users(): """List handles in the cache, with engagement counts as a sort hint.""" if not db_path.exists(): return jsonify({"users": []}) db = sqlite3.connect(db_path) db.row_factory = sqlite3.Row rows = db.execute( """SELECT p.handle, p.did, p.followers_count, p.posts_count, (SELECT COUNT(*) FROM posts WHERE author_did = p.did) AS cached_posts FROM profiles p WHERE p.posts_count IS NOT NULL ORDER BY cached_posts DESC, p.followers_count DESC NULLS LAST""" ).fetchall() return jsonify({"users": [dict(r) for r in rows]})
@app.post("/api/world_data") @rate_limit(30) def post_world_data(): """Return raw event stream + user crops for client-side sim.
This is the data plane: cheap, denormalized, cacheable. The sim itself runs in the browser now — server only serves the inputs. """ body = request.get_json(silent=True) or {} handle = body.get("handle") or "cee.wtf" if not valid_handle(handle): return jsonify({"error": "invalid handle"}), 400 top = clamp_int(body.get("top_neighbors", 24), 0, 200, 24) max_age_days = body.get("max_age_days") if max_age_days is not None: max_age_days = clamp_int(max_age_days, 1, 3650, 365) neighbor_dids_in = clean_did_list(body.get("neighbor_dids"))
if not db_path.exists(): return jsonify({"error": "no cache"}), 404 db = sqlite3.connect(db_path) db.row_factory = sqlite3.Row row = db.execute("SELECT did FROM profiles WHERE handle = ? AND did <> ''", (handle,)).fetchone() if not row or not row["did"]: return jsonify({"error": "not in cache"}), 404 focus_did = row["did"]
from sim.world import neighbor_query, NeighborInfo if neighbor_dids_in is not None: pool = neighbor_query(db, focus_did, 1000) wanted = set(neighbor_dids_in) by_did = {n.did: n for n in pool} neighbors = [] for did in neighbor_dids_in: if did == focus_did: continue if did in by_did: neighbors.append(by_did[did]) else: r = db.execute("SELECT handle FROM profiles WHERE did = ?", (did,)).fetchone() if r and r["handle"]: neighbors.append(NeighborInfo( did=did, handle=r["handle"], social_class="other", engagement_count=0, origin=(0, 0), )) else: neighbors = neighbor_query(db, focus_did, top)
from sim.crops import crop_for_user from sim.replay import load_events_bulk sim_dids = [focus_did] + [n.did for n in neighbors]
users = [{"did": focus_did, "handle": handle, "crop": crop_for_user(focus_did)}] for n in neighbors: users.append({"did": n.did, "handle": n.handle, "crop": crop_for_user(n.did)})
# Single SQL for all farms — already time-sorted. events = [list(row) for row in load_events_bulk(db, sim_dids, max_age_days=max_age_days)]
return fast_json({ "focus_did": focus_did, "focus_handle": handle, "users": users, "neighbors": [ { "did": n.did, "handle": n.handle, "social_class": n.social_class, "engagement_count": n.engagement_count, } for n in neighbors ], "events": events, })
return app
def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--db", default=str(DEFAULT_DB)) ap.add_argument("--host", default="0.0.0.0") ap.add_argument("--port", type=int, default=8920) args = ap.parse_args()
app = create_app(Path(args.db)) app.run(host=args.host, port=args.port, debug=False, use_reloader=False) return 0
if __name__ == "__main__": raise SystemExit(main())