# Bitable Change Notification — Implementation Guide **Date:** 2026-07-01 **Status:** Ready for implementation --- ## TL;DR Extend `feishu-app` (a separate Python/FastAPI service deployed via Docker, repo at `../../aptsell/feishu-app`) to listen for `drive.file.bitable_record_changed_v1` WebSocket events, maintain a local cache of `{record_id → assignee}`, and push change notifications to Bugger via **Server-Sent Events (SSE)**. Bugger opens an outbound SSE connection to feishu-app — no local HTTP server required. Key design decisions: - **SSE (Bugger → feishu-app, outbound)**: Bugger initiates the connection, so it works through NAT. feishu-app pushes events down the stream. - **Assignee-aware routing**: feishu-app caches `{record_id → assignee}` and only notifies Bugger instances whose assignee actually changed. Eliminates redundant API calls. - **Cache reconciliation**: feishu-app does a full Bitable pull on startup and daily to keep the cache accurate. - **Bugger `fetchNow()` unchanged**: zero signature changes to the existing polling pipeline. - **Feature flag:** `BITABLE_CHANGE_NOTIFICATION_ENABLED=false` by default. When disabled, **zero resources are allocated** — no imports, no Redis connections, no SSE endpoint, no WebSocket handler registration. - **Code to write:** ~420 LOC Python (feishu-app) + ~120 LOC Swift (Bugger) - **New dependencies:** None — `lark-oapi==1.4.20` and `httpx` already in requirements.txt - **Bugger config change:** Add `feishuAppBaseURL` field to `AppConfig` - **Estimated effort:** 2 days implementation + configuration --- ## 1. How It Works ``` ┌──────────────────────┐ WebSocket push ┌──────────────────┐ │ Feishu Cloud │ ────bitable_record─────▶ │ feishu-app │ │ (Bitable changed) │ changed_v1 │ (Docker) │ └──────────────────────┘ └────────┬─────────┘ │ ① Fetch changed record (1 API call) ② Diff cached assignee ↔ current ③ Only notify if assignee changed │ SSE push to matching assignee's Bugger instance(s) │ ┌──────────────────────┐ │ │ Bugger (User A) │◀═══ SSE: assignee changed ═══════┘ │ menu bar app │ │ │──▶ fetchNow() (existing, unchanged) │ PollerService │──▶ Feishu Bitable API (existing flow) │ (periodic fallback) │ └──────────────────────┘ ``` 1. User edits a Bitable record (any field) 2. Feishu pushes `drive.file.bitable_record_changed_v1` via WebSocket to `feishu-app` 3. `feishu-app` fetches the changed record (1 API call), compares `current_assignee` against its local cache 4. If assignee unchanged (e.g. priority edit) → **skip, no notification**. 5. If assignee changed (new bug, reassignment, deletion) → push wake-up signal via SSE to the affected Bugger instance(s) 6. Bugger triggers its existing `fetchNow()` → `FeishuService.fetchBugs(assigneeName:)` (which already filters by assignee) 7. Bugger's `PollerService` remains as periodic fallback (15-30 min) **API call economics (team of 20 developers):** | Scenario | Dumb relay (old design) | Assignee-aware (this design) | |----------|------------------------|------------------------------| | Tester edits priority | 20 Bugger fetches | 1 record fetch → skip | | Assignee changed A→B | 20 Bugger fetches | 1 record fetch + 2 Bugger fetches (A's + B's) | | New bug assigned to A | 20 Bugger fetches | 1 record fetch + 1 Bugger fetch (A's) | --- ## 2. Prerequisites — Feishu Open Platform Configuration ### 2.1 Verify the App `feishu-app` is a separate repository at `../../aptsell/feishu-app/feishu-service-python/`. It uses: - `FEISHU_APP_ID=cli_a7245c3f17745013` (from `.env.example`) - `FEISHU_APP_SECRET` (configured in deployment) - `VERIFICATION_TOKEN` (configured in deployment) - SDK: `lark-oapi==1.4.20` (confirmed in `requirements.txt`) - Existing Redis for dedup/session storage Confirm the `P2DriveFileBitableRecordChangedV1` model and `register_p2_drive_file_bitable_record_changed_v1` builder method are available in the SDK version. If not, bump to `lark-oapi>=1.5.0`. ### 2.2 Add Required Permission Scope The app needs **one** of the following scopes to call the subscribe API and fetch records: | Scope | Description | |-------|-------------| | `drive:drive` | Full cloud document access (recommended — covers subscribe + record read) | | `bitable:app` | Bitable read/write access | **Steps:** 1. Go to [Feishu Open Platform](https://open.feishu.cn/) → Apps → your app 2. **Permissions** (权限管理) → **Add Permission** (添加权限) 3. Search for `bitable:app` (or `drive:drive`) 4. Add the permission 5. **Publish** a new app version for the permission change to take effect > ⚠️ **Important:** Feishu requires creating and publishing a new app version after any permission change. Without this, the subscribe API returns 403. ### 2.3 Add App to Each Bitable For each Bitable that Bugger should receive real-time events for: 1. Open the Bitable in Feishu 2. Click **"..."** (top-right menu) → **"更多"** → **"添加文档应用"** 3. Select the feishu-app from the list 4. Grant it **"可管理"** (manage) permission This is a **one-time setup per Bitable**. Once added, the app can subscribe and receive events regardless of individual users' roles on that Bitable. --- ## 2A. Feature Flag The entire Bitable change notification system is gated behind a single boolean flag. When disabled, **no resources are consumed** — no imports, no class instances, no Redis keys, no API calls. ### 2A.1 Setting Definition Add to `src/app/config.py` (follows the existing `enable_link_summary` pattern, around line 76): ```python # Bitable change notification feature flag (disabled by default) bitable_change_notification_enabled: bool = ( os.getenv("BITABLE_CHANGE_NOTIFICATION_ENABLED", "false").lower() == "true" ) ``` ### 2A.2 `.env` Entry ``` # Enable Bitable change notification (SSE push to Bugger) BITABLE_CHANGE_NOTIFICATION_ENABLED=false ``` ### 2A.3 What the Flag Gates ``` BITABLE_CHANGE_NOTIFICATION_ENABLED │ ├─ false (default) │ ├─ No import of BitableEventService │ ├─ No import of bitable_subscription router │ ├─ No import of P2DriveFileBitableRecordChangedV1 │ ├─ No WebSocket handler registration │ ├─ No SSE endpoint registered on FastAPI │ ├─ No Redis cache keys created │ ├─ No full Bitable pull on startup │ ├─ No reconciliation task │ └─ No Feishu subscribe API calls │ └─ true ├─ Import BitableEventService (lazy, only on first SSE connection) ├─ Register SSE router ├─ Register WebSocket handler ├─ Redis cache populated on first use ├─ Full pull + reconciliation task └─ Feishu subscribe on first Bugger connection ``` ### 2A.4 Gating Pattern Every change point uses the same guard — no dead imports, no orphaned constants, no half-initialized state: ```python if settings.bitable_change_notification_enabled: # ... register handler, import service, wire router, start reconciliation ``` This follows the existing `if settings.platform.lower() == "dingtalk":` branching pattern already used in `main.py`. --- ## 3. Code Changes — feishu-app (Python) Repository: `../../aptsell/feishu-app/feishu-service-python/` ### 3.1 New File: `src/app/services/bitable_event_service.py` The core service — manages the assignee cache, handles WebSocket events, and serves SSE connections. ```python """ Bitable change event service. Maintains {record_id → assignee} cache, filters events by assignee transitions, and pushes notifications to Bugger instances via SSE. """ import asyncio import json import logging import time from typing import Dict, Any, Optional, Set, List from dataclasses import dataclass, field import httpx from src.app.config import settings logger = logging.getLogger(__name__) @dataclass class BuggerSSEConnection: """An SSE connection from a Bugger instance.""" queue: asyncio.Queue file_token: str assignee_field: str # e.g. "负责人", "assignee" assignee_name: str # e.g. "张三" class BitableEventService: """ Manages Bitable event subscriptions, assignee cache, and SSE fan-out. Lifecycle: 1. bugger opens SSE: GET /api/v1/bitable/events?file_token=...&assignee_field=...&assignee_name=... 2. First connection for a file_token → feishu-app calls Feishu subscribe API → triggers initial full pull to warm {record_id → assignee} cache 3. WebSocket events arrive → fetch changed record → diff assignee → push to matching SSE connections 4. Daily cron: full pull → reconcile cache → notify missed transitions 5. Last connection for a file_token → unsubscribe from Feishu """ def __init__(self, bot_service): self._bot = bot_service # file_token → assignee_field → record_id → assignee_value self._cache: Dict[str, Dict[str, Dict[str, str]]] = {} # SSE connections, keyed by (file_token, assignee_name) self._connections: Dict[str, List[BuggerSSEConnection]] = {} # Track Feishu subscription state per file_token self._feishu_subscribed: Set[str] = set() # Reconciliation task self._reconcile_task: Optional[asyncio.Task] = None # ── Feishu subscribe / unsubscribe (internal) ──────────── async def _subscribe_to_file(self, file_token: str) -> bool: """Call Feishu subscribe API to start receiving WebSocket events.""" try: headers = await self._bot._get_headers() url = (f"{self._bot.base_url}/drive/v1/files/" f"{file_token}/subscribe?file_type=bitable") async with httpx.AsyncClient() as client: response = await client.post(url, headers=headers, timeout=10.0) result = response.json() if result.get("code") == 0: logger.info(f"✅ Subscribed to Feishu events for {file_token}") self._feishu_subscribed.add(file_token) return True elif result.get("code") == 1069603: logger.error( f"❌ Permission denied for {file_token}. " f"App must be added as document manager. " f"Feishu msg: {result.get('msg')}" ) return False else: logger.error(f"Subscribe failed for {file_token}: {result}") return False except Exception as e: logger.error(f"Subscribe exception for {file_token}: {e}") return False async def _unsubscribe_from_file(self, file_token: str) -> None: """Call Feishu unsubscribe API.""" try: headers = await self._bot._get_headers() url = (f"{self._bot.base_url}/drive/v1/files/" f"{file_token}/unsubscribe?file_type=bitable") async with httpx.AsyncClient() as client: await client.post(url, headers=headers, timeout=10.0) self._feishu_subscribed.discard(file_token) logger.info(f"Unsubscribed from Feishu events for {file_token}") except Exception as e: logger.error(f"Unsubscribe exception for {file_token}: {e}") # ── Assignee cache ────────────────────────────────────── async def warm_cache(self, file_token: str, assignee_field: str) -> Dict[str, str]: """ Full pull of Bitable records to build/refresh the assignee cache. Returns the built cache slice: {record_id: assignee_value}. """ cache: Dict[str, str] = {} page_token: Optional[str] = None headers = await self._bot._get_headers() try: while True: url = ( f"{self._bot.base_url}/bitable/v1/apps/{file_token}" f"/tables/tblXXXXXXXXXXXXXXXX/records" f"?page_size=500" ) if page_token: url += f"&page_token={page_token}" async with httpx.AsyncClient() as client: response = await client.get(url, headers=headers, timeout=30.0) data = response.json() if data.get("code") != 0: logger.error(f"Full pull failed for {file_token}: {data}") break items = (data.get("data") or {}).get("items") or [] for item in items: record_id = item.get("record_id") fields = item.get("fields") or {} assignee = self._extract_assignee(fields, assignee_field) if record_id and assignee: cache[record_id] = assignee if not data.get("data", {}).get("has_more"): break page_token = data.get("data", {}).get("page_token") # Store in cache structure: file_token → assignee_field → record_id → assignee file_cache = self._cache.setdefault(file_token, {}) file_cache[assignee_field] = cache logger.info( f"Cache warmed for {file_token} " f"(field={assignee_field}): {len(cache)} records" ) return cache except Exception as e: logger.error(f"Cache warm exception for {file_token}: {e}", exc_info=True) return cache def _extract_assignee(self, fields: Dict[str, Any], assignee_field: str) -> Optional[str]: """ Extract assignee name from a record's fields. Handles Feishu's user-type field format: - Single user: {"name": "张三", "id": "ou_xxx", ...} - List of users: [{"name": "张三", ...}] - String/text field: "张三" """ value = fields.get(assignee_field) if value is None: return None if isinstance(value, str): return value if isinstance(value, dict): return value.get("name") or value.get("id") if isinstance(value, list) and len(value) > 0: first = value[0] if isinstance(first, dict): return first.get("name") or first.get("id") if isinstance(first, str): return first return None def _cache_key(self, file_token: str, assignee_name: str) -> str: return f"{file_token}:{assignee_name}" # ── SSE connection management ─────────────────────────── async def add_connection(self, conn: BuggerSSEConnection) -> bool: """ Add an SSE connection. Returns True if the cache was ready, False if a warm is needed. """ key = self._cache_key(conn.file_token, conn.assignee_name) self._connections.setdefault(key, []).append(conn) logger.info( f"SSE connection added: file={conn.file_token}, " f"assignee={conn.assignee_name}, total={len(self._connections[key])}" ) # First connection for this file_token? Subscribe and warm cache if conn.file_token not in self._feishu_subscribed: await self._subscribe_to_file(conn.file_token) cache_ready = ( conn.file_token in self._cache and conn.assignee_field in self._cache[conn.file_token] ) return cache_ready def remove_connection(self, conn: BuggerSSEConnection) -> None: """Remove an SSE connection. Unsubscribe if last connection.""" key = self._cache_key(conn.file_token, conn.assignee_name) bucket = self._connections.get(key, []) if conn in bucket: bucket.remove(conn) if not bucket: self._connections.pop(key, None) # Last connection for this file_token? Unsubscribe has_other = any( c.file_token == conn.file_token for bucket in self._connections.values() for c in bucket ) if not has_other and conn.file_token in self._feishu_subscribed: asyncio.create_task(self._unsubscribe_from_file(conn.file_token)) async def notify_connections( self, file_token: str, assignee_name: str, reason: str = "changed" ) -> int: """Push a wake-up event to all SSE connections matching the assignee.""" key = self._cache_key(file_token, assignee_name) notified = 0 for conn in self._connections.get(key, []): try: await conn.queue.put({ "event": "change", "data": json.dumps({ "file_token": file_token, "assignee_name": assignee_name, "reason": reason, "timestamp": int(time.time()), }) }) notified += 1 except Exception as e: logger.error(f"Failed to queue SSE event: {e}") if notified: logger.info(f"Notified {notified} Bugger instance(s) for {assignee_name}") return notified # ── Event handling ────────────────────────────────────── async def handle_bitable_record_changed(self, data) -> None: """ Handle a bitable_record_changed_v1 event from WebSocket. Fetch changed record → diff assignee → notify only if assignee changed. """ try: header = getattr(data, 'header', None) event = getattr(data, 'event', None) event_id = getattr(header, 'event_id', None) file_token = getattr(event, 'file_token', None) table_id = getattr(event, 'table_id', None) revision = getattr(event, 'revision', None) logger.info( f"🔔 Bitable change event: event_id={event_id}, " f"file_token={file_token}, table_id={table_id}, revision={revision}" ) if not file_token or file_token not in self._feishu_subscribed: return # Dedup via Redis if event_id: redis = await self._bot._get_redis_client() dedup_key = f"feishu:bitable_event:{event_id}" already_seen = await redis.get(dedup_key) if already_seen: logger.debug(f"Dedup: skipping duplicate event {event_id}") return await redis.set(dedup_key, "1", ex=3600) # Process each changed record action_list = getattr(event, 'action_list', None) or [] headers = await self._bot._get_headers() for action in action_list: record_id = getattr(action, 'record_id', None) act_type = getattr(action, 'action', 'unknown') if not record_id: continue # Fetch the full record record = await self._fetch_record( file_token, table_id, record_id, headers ) if record is None: # Record deleted — notify old assignee(s) for field_name, assignee_cache in ( self._cache.get(file_token, {}).items() ): old_assignee = assignee_cache.pop(record_id, None) if old_assignee: logger.info( f"Record {record_id} deleted, " f"notifying old assignee={old_assignee}" ) await self.notify_connections( file_token, old_assignee, reason="deleted" ) continue fields = record.get("fields") or {} # Check each tracked assignee field for this file_token file_cache = self._cache.setdefault(file_token, {}) for assignee_field in list(file_cache.keys()): current = self._extract_assignee(fields, assignee_field) cached = file_cache[assignee_field].get(record_id) if current == cached: logger.debug( f"Record {record_id}: assignee unchanged ({current}) — skip" ) continue # Assignee changed — update cache and notify logger.info( f"Record {record_id}: assignee " f"{cached or '(none)'} → {current or '(none)'}" ) if current: file_cache[assignee_field][record_id] = current else: file_cache[assignee_field].pop(record_id, None) # Notify new assignee if current: await self.notify_connections( file_token, current, reason="assigned" ) # Notify old assignee (bug moved away) if cached and cached != current: await self.notify_connections( file_token, cached, reason="unassigned" ) except Exception as e: logger.error(f"Error handling bitable change event: {e}", exc_info=True) async def _fetch_record( self, file_token: str, table_id: str, record_id: str, headers: Dict[str, str] ) -> Optional[Dict[str, Any]]: """Fetch a single Bitable record. Returns None if deleted.""" try: url = ( f"{self._bot.base_url}/bitable/v1/apps/{file_token}" f"/tables/{table_id}/records/{record_id}" ) async with httpx.AsyncClient() as client: response = await client.get(url, headers=headers, timeout=10.0) data = response.json() if data.get("code") == 0: return (data.get("data") or {}).get("record") elif data.get("code") == 172001: # record not found / deleted return None else: logger.warning(f"Failed to fetch record {record_id}: {data}") return None except Exception as e: logger.error(f"Fetch record exception for {record_id}: {e}") return None # ── Daily reconciliation ──────────────────────────────── async def start_reconciliation(self, interval_hours: int = 24): """Start periodic full-pull reconciliation.""" if self._reconcile_task: return async def _run(): while True: await asyncio.sleep(interval_hours * 3600) await self._reconcile_all() self._reconcile_task = asyncio.create_task(_run()) logger.info(f"Reconciliation task started (every {interval_hours}h)") async def _reconcile_all(self): """Full pull for all active file_tokens, reconcile cache.""" logger.info("Starting daily reconciliation...") for file_token, field_caches in self._cache.items(): for assignee_field, old_cache in field_caches.items(): try: new_cache = await self.warm_cache(file_token, assignee_field) # Find and notify transitions all_ids = set(old_cache.keys()) | set(new_cache.keys()) for record_id in all_ids: old_val = old_cache.get(record_id) new_val = new_cache.get(record_id) if old_val != new_val: logger.info( f"Reconciliation: record {record_id} " f"{old_val or '(none)'} → {new_val or '(none)'}" ) if new_val: await self.notify_connections( file_token, new_val, reason="reconciled" ) if old_val: await self.notify_connections( file_token, old_val, reason="reconciled" ) # Update cache in place self._cache[file_token][assignee_field] = new_cache except Exception as e: logger.error( f"Reconciliation failed for {file_token}/{assignee_field}: {e}", exc_info=True ) logger.info("Daily reconciliation complete") ``` ### 3.2 New File: `src/app/api/endpoints/bitable_subscription.py` The SSE endpoint that Bugger connects to: ```python """ SSE endpoint for Bitable change notifications. Bugger instances connect here to receive real-time wake-up signals. """ import json import asyncio import logging from typing import Optional from fastapi import APIRouter, Query, Request from fastapi.responses import StreamingResponse logger = logging.getLogger(__name__) router = APIRouter() # Will be set by main.py on startup bitable_event_service = None @router.get("/events") async def bitable_events( request: Request, file_token: str = Query(..., description="Bitable file token"), assignee_field: str = Query(..., description="Field name for assignee"), assignee_name: str = Query(..., description="This Bugger's assignee name"), table_id: str = Query("", description="Table ID (for record fetch URL)"), ): """ SSE endpoint. Bugger opens a long-lived connection here. feishu-app pushes change events only when the assignee matches. """ if not bitable_event_service: return StreamingResponse( _error_stream("Bitable service not initialized"), media_type="text/event-stream" ) from src.app.services.bitable_event_service import BuggerSSEConnection conn = BuggerSSEConnection( queue=asyncio.Queue(), file_token=file_token, assignee_field=assignee_field, assignee_name=assignee_name, ) # Register connection service = bitable_event_service cache_ready = await service.add_connection(conn) # If cache not ready for this file_token/field, warm it if not cache_ready: logger.info(f"Cache cold for {file_token}/{assignee_field}, warming...") await service.warm_cache(file_token, assignee_field) else: logger.info(f"Cache already warm for {file_token}/{assignee_field}") logger.info( f"Bugger SSE connected: file={file_token}, " f"assignee={assignee_name}, field={assignee_field}" ) async def event_stream(): try: # Send initial connected event yield f"event: connected\ndata: {json.dumps({'status': 'ok'})}\n\n" while True: # Check client disconnect if await request.is_disconnected(): break try: # Wait for events with heartbeat timeout event = await asyncio.wait_for(conn.queue.get(), timeout=30.0) yield f"event: {event['event']}\ndata: {event['data']}\n\n" except asyncio.TimeoutError: # Heartbeat yield ": heartbeat\n\n" except asyncio.CancelledError: pass finally: service.remove_connection(conn) logger.info( f"Bugger SSE disconnected: file={file_token}, " f"assignee={assignee_name}" ) return StreamingResponse( event_stream(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", # disable nginx buffering } ) async def _error_stream(message: str): yield f"event: error\ndata: {json.dumps({'error': message})}\n\n" ``` ### 3.3 Modify `run_feishu_service.py` All changes are gated behind the feature flag. When disabled, the file is untouched at runtime. **Add import (after the existing `lark_oapi` imports, around line 10):** ```python if settings.bitable_change_notification_enabled: from lark_oapi.api.drive.v1.model.p2_drive_file_bitable_record_changed_v1 import ( P2DriveFileBitableRecordChangedV1, ) ``` **Add import for the new service (after the existing service imports, around line 16):** ```python if settings.bitable_change_notification_enabled: from src.app.services.bitable_event_service import BitableEventService ``` **Add the event handler function (after `on_p2_card_action_trigger`, around line 302):** ```python # Bitble change notification — only defined when feature is enabled if settings.bitable_change_notification_enabled: bitable_event_service = None def on_p2_drive_file_bitable_record_changed_v1( data: P2DriveFileBitableRecordChangedV1 ): """Handle Bitable record change events from WebSocket.""" try: logger.info("=" * 60) logger.info("🔔 收到 Bitable 记录变更事件!") evt = getattr(data, 'event', None) header = getattr(data, 'header', None) logger.info(f"事件ID: {getattr(header, 'event_id', None)}") logger.info(f"文件Token: {getattr(evt, 'file_token', None)}") logger.info(f"表格ID: {getattr(evt, 'table_id', None)}") logger.info(f"版本: {getattr(evt, 'revision', None)}") action_list = getattr(evt, 'action_list', None) if action_list: for action in action_list: logger.info( f" Action: {getattr(action, 'action', 'unknown')} " f"on record {getattr(action, 'record_id', None)}" ) # Process asynchronously if bitable_event_service: async def _handle(): await bitable_event_service.handle_bitable_record_changed(data) task = asyncio.create_task(_handle()) def handle_exception(t): try: t.result() except Exception as e: logger.error(f"Bitable事件处理异常: {e}", exc_info=True) task.add_done_callback(handle_exception) logger.info("=" * 60) except Exception as e: logger.error(f"处理 Bitable 变更事件出错: {e}", exc_info=True) else: # Feature disabled — no imports, no handler, no service reference bitable_event_service = None ``` **Register the handler in `FeishuStreamManager.start()` (add after the card action registration block, around line 349):** ```python # 注册 Bitable 记录变更处理器(仅在启用时) if settings.bitable_change_notification_enabled: if hasattr(builder, "register_p2_drive_file_bitable_record_changed_v1"): builder = builder.register_p2_drive_file_bitable_record_changed_v1( on_p2_drive_file_bitable_record_changed_v1 ) logger.info("已注册 bitable_record_changed 处理器") else: logger.warning( "当前 SDK 版本不支持 register_p2_drive_file_bitable_record_changed_v1" ) ``` **Initialize the service in `main()` (add after `bot_service = FeishuBotService()`, around line 426):** ```python # 初始化 Bitable 事件服务(仅在启用时) if settings.bitable_change_notification_enabled: global bitable_event_service bitable_event_service = BitableEventService(bot_service) logger.info("✅ Bitable 事件服务已初始化") ``` ### 3.4 Modify `src/app/main.py` Register the SSE endpoint and wire the service — all gated behind the feature flag. **Add after existing router registrations (around line 94):** ```python # Register Bitable SSE endpoint (Bugger connects to this) — only when enabled if settings.bitable_change_notification_enabled: from src.app.api.endpoints import bitable_subscription app.include_router( bitable_subscription.router, prefix="/api/v1/bitable", tags=["bitable"] ) ``` **Wire the service instance (add in `lifespan` startup, after `stream_manager.start()`):** ```python # Wire bitable_event_service into the SSE router (only when enabled) if settings.bitable_change_notification_enabled: from run_feishu_service import bitable_event_service as global_svc if global_svc: bitable_subscription.bitable_event_service = global_svc # Start daily reconciliation task await global_svc.start_reconciliation(interval_hours=24) logger.info("✅ Bitable event service wired and reconciliation started") ``` --- ## 4. Code Changes — Bugger (Swift / macOS) Repository: this project (`bugger/`) ### 4.1 New File: `Sources/Services/BitableEventService.swift` An SSE client that connects to feishu-app and triggers `fetchNow()` on change events. ```swift import Foundation /// Connects to feishu-app via SSE to receive Bitable change notifications. /// Triggers PollerService.fetchNow() when the assignee changes. final class BitableEventService { static let shared = BitableEventService() private var task: URLSessionDataTask? private var session: URLSession? private var isConnected = false private init() {} // MARK: - Public /// Start the SSE connection. Safe to call multiple times — no-op if already connected. func connect() { guard !isConnected else { return } guard let url = buildURL() else { BuggerLog.info("BitableEventService: feishuAppBaseURL not configured, skipping SSE") return } isConnected = true BuggerLog.info("BitableEventService: connecting to \(url.absoluteString)") let session = URLSession( configuration: .default, delegate: SSEDelegate(), delegateQueue: nil ) self.session = session var request = URLRequest(url: url) request.setValue("text/event-stream", forHTTPHeaderField: "Accept") request.timeoutInterval = TimeInterval(INT_MAX) // no timeout task = session.dataTask(with: request) task?.resume() } /// Disconnect the SSE stream. func disconnect() { task?.cancel() task = nil session?.invalidateAndCancel() session = nil isConnected = false BuggerLog.info("BitableEventService: disconnected") } // MARK: - Private private func buildURL() -> URL? { guard let config = AppStateService.shared.config, config.isConfigured, !config.feishuAppBaseURL.isEmpty else { return nil } let assigneeName = config.assigneeName ?? TokenManager.shared.persistedUserInfo?.name ?? "" let assigneeField = config.fieldMappings.assignee guard !assigneeName.isEmpty, !assigneeField.isEmpty else { BuggerLog.warning("BitableEventService: assignee name or field not configured") return nil } var components = URLComponents( string: "\(config.feishuAppBaseURL)/api/v1/bitable/events" )! components.queryItems = [ URLQueryItem(name: "file_token", value: config.appToken), URLQueryItem(name: "table_id", value: config.tableId), URLQueryItem(name: "assignee_field", value: assigneeField), URLQueryItem(name: "assignee_name", value: assigneeName), ] return components.url } } // MARK: - URLSessionDataDelegate (SSE parsing) private final class SSEDelegate: NSObject, URLSessionDataDelegate { private var buffer = "" func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive data: Data) { guard let chunk = String(data: data, encoding: .utf8) else { return } buffer.append(chunk) // Parse SSE frames: lines ending with \n\n while let range = buffer.range(of: "\n\n") { let frame = String(buffer[..