#!/usr/bin/env python3 """Opt-in sync of local Zed Hours JSONL logs to a Zed Hours server. The command sends only records that have not been acknowledged locally. Every outgoing record has a deterministic, device-scoped source_id, so replaying a batch after a network failure is safe when the server enforces source_id uniqueness. """ import argparse import hashlib import json import os import subprocess import sys from pathlib import Path from typing import Any from urllib import error, request HEARTBEATS_ENDPOINT = "/api/v1/sync/zed-heartbeats" COMMITS_ENDPOINT = "/api/v1/sync/zed-commits" STATE_FILE_NAME = "sync-state.json" DEFAULT_BATCH_SIZE = 100 def get_project_root(project_root: Path | None) -> Path: """Resolve an explicit root, then use the Git root or current directory.""" if project_root is not None: return project_root.resolve() try: root = subprocess.check_output( ["git", "rev-parse", "--show-toplevel"], text=True, stderr=subprocess.DEVNULL, ).strip() return Path(root).resolve() except (subprocess.CalledProcessError, FileNotFoundError): return Path.cwd().resolve() def load_state(state_path: Path) -> dict[str, Any]: if not state_path.exists(): return {"version": 1, "files": {}} try: state = json.loads(state_path.read_text(encoding="utf-8")) if isinstance(state, dict) and isinstance(state.get("files"), dict): return state except (OSError, json.JSONDecodeError): pass print(f"Warning: ignoring invalid sync state: {state_path}", file=sys.stderr) return {"version": 1, "files": {}} def save_state(state_path: Path, state: dict[str, Any]) -> None: temporary_path = state_path.with_suffix(".tmp") temporary_path.write_text(json.dumps(state, indent=2, sort_keys=True) + "\n", encoding="utf-8") temporary_path.replace(state_path) def source_id(device_id: str, record: dict[str, Any]) -> str: """Return a stable ID derived from the device and canonical record content.""" content = json.dumps(record, ensure_ascii=False, sort_keys=True, separators=(",", ":")) return hashlib.sha256(f"{device_id}\0{content}".encode()).hexdigest() def outgoing_record(record: dict[str, Any], device_id: str) -> dict[str, Any]: """Remove local absolute-path fields and attach the idempotency key.""" sanitized = {key: value for key, value in record.items() if key not in {"project", "repo", "source_id"}} sanitized["source_id"] = source_id(device_id, sanitized) return sanitized def read_batch(log_path: Path, offset: int, batch_size: int, device_id: str) -> tuple[list[dict[str, Any]], int]: """Read one batch and its final byte offset, skipping malformed JSON lines.""" records: list[dict[str, Any]] = [] next_offset = offset with log_path.open("rb") as log_file: log_file.seek(offset) while len(records) < batch_size: line = log_file.readline() if not line: break next_offset = log_file.tell() try: parsed = json.loads(line.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): print(f"Warning: skipping malformed JSONL record in {log_path}", file=sys.stderr) continue if not isinstance(parsed, dict): print(f"Warning: skipping non-object JSONL record in {log_path}", file=sys.stderr) continue records.append(outgoing_record(parsed, device_id)) return records, next_offset def post_records(server_url: str, endpoint: str, token: str, device_id: str, project_slug: str, records: list[dict[str, Any]]) -> None: payload = json.dumps( {"device_id": device_id, "project_slug": project_slug, "records": records}, ensure_ascii=False, ).encode("utf-8") sync_request = request.Request( f"{server_url.rstrip('/')}{endpoint}", data=payload, headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", "Accept": "application/json", }, method="POST", ) try: with request.urlopen(sync_request, timeout=30) as response: if not 200 <= response.status < 300: raise RuntimeError(f"server returned HTTP {response.status}") except error.HTTPError as exc: response_body = exc.read().decode("utf-8", errors="replace").strip() detail = f": {response_body}" if response_body else "" raise RuntimeError(f"server returned HTTP {exc.code}{detail}") from exc except error.URLError as exc: raise RuntimeError(f"could not reach sync server: {exc.reason}") from exc def sync_log( log_path: Path, endpoint: str, state: dict[str, Any], state_path: Path, server_url: str, token: str, device_id: str, project_slug: str, batch_size: int, ) -> int: if not log_path.exists(): print(f"No log file: {log_path}") return 0 files_state = state["files"] file_key = log_path.name offset = files_state.get(file_key, {}).get("offset", 0) if not isinstance(offset, int) or offset < 0 or offset > log_path.stat().st_size: offset = 0 synced = 0 while True: records, next_offset = read_batch(log_path, offset, batch_size, device_id) if not records: if next_offset != offset: files_state[file_key] = {"offset": next_offset} save_state(state_path, state) return synced post_records(server_url, endpoint, token, device_id, project_slug, records) files_state[file_key] = {"offset": next_offset} save_state(state_path, state) synced += len(records) offset = next_offset def required_setting(name: str) -> str: value = os.environ.get(name, "").strip() if not value: raise ValueError(f"{name} must be set") return value def main() -> int: parser = argparse.ArgumentParser(description="Sync local Zed Hours logs to a server.") parser.add_argument("--project-root", type=Path, default=None, help="Project root directory") parser.add_argument("--project-slug", default=None, help="Server project identifier (defaults to project root name)") parser.add_argument("--batch-size", type=int, default=DEFAULT_BATCH_SIZE, help="Records per request (default: 100)") args = parser.parse_args() if args.batch_size < 1: parser.error("--batch-size must be at least 1") try: server_url = required_setting("ZED_HOURS_SERVER_URL") token = required_setting("ZED_HOURS_SERVER_TOKEN") device_id = required_setting("ZED_HOURS_DEVICE_ID") except ValueError as exc: parser.error(str(exc)) project_root = get_project_root(args.project_root) project_slug = args.project_slug or project_root.name if not project_slug: parser.error("--project-slug must not be empty") log_dir = project_root / ".zed-hours" state_path = log_dir / STATE_FILE_NAME state = load_state(state_path) try: heartbeat_count = sync_log( log_dir / "heartbeats.jsonl", HEARTBEATS_ENDPOINT, state, state_path, server_url, token, device_id, project_slug, args.batch_size, ) commit_count = sync_log( log_dir / "commits.jsonl", COMMITS_ENDPOINT, state, state_path, server_url, token, device_id, project_slug, args.batch_size, ) except (OSError, RuntimeError) as exc: print(f"Sync failed: {exc}", file=sys.stderr) return 1 print(f"Synced {heartbeat_count} heartbeat(s) and {commit_count} commit(s) for project '{project_slug}'.") return 0 if __name__ == "__main__": raise SystemExit(main())