48 KiB

Raw Blame History

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)  │
└──────────────────────┘

User edits a Bitable record (any field)
Feishu pushes drive.file.bitable_record_changed_v1 via WebSocket to feishu-app
feishu-app fetches the changed record (1 API call), compares current_assignee against its local cache
If assignee unchanged (e.g. priority edit) → skip, no notification.
If assignee changed (new bug, reassignment, deletion) → push wake-up signal via SSE to the affected Bugger instance(s)
Bugger triggers its existing fetchNow() → FeishuService.fetchBugs(assigneeName:) (which already filters by assignee)
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:

Go to Feishu Open Platform → Apps → your app
Permissions (权限管理) → Add Permission (添加权限)
Search for bitable:app (or drive:drive)
Add the permission
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:

Open the Bitable in Feishu
Click "..." (top-right menu) → "更多" → "添加文档应用"
Select the feishu-app from the list
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):

# 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:

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.

"""
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:

"""
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):

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):

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):

# 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):

            # 注册 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):

    # 初始化 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`

Add after existing router registrations (around line 94):

# 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()):

    # 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.

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[..<range.lowerBound])
            buffer.removeSubrange(..<range.upperBound)
            processFrame(frame)
        }
    }

    func urlSession(_ session: URLSession, task: URLSessionTask,
                    didCompleteWithError error: Error?) {
        if let error = error {
            BuggerLog.warning("BitableEventService SSE error: \(error.localizedDescription)")
        }
        BitableEventService.shared.disconnect()

        // Auto-reconnect after delay
        DispatchQueue.main.asyncAfter(deadline: .now() + 30) {
            BuggerLog.info("BitableEventService: attempting reconnect...")
            BitableEventService.shared.connect()
        }
    }

    private func processFrame(_ frame: String) {
        var eventType = ""
        var data = ""

        for line in frame.components(separatedBy: "\n") {
            if line.hasPrefix("event: ") {
                eventType = String(line.dropFirst(7))
            } else if line.hasPrefix("data: ") {
                data = String(line.dropFirst(6))
            }
        }

        switch eventType {
        case "change":
            BuggerLog.info("🔔 Bitable change push received, triggering fetchNow()")
            Task { await PollerService.shared.fetchNow() }

        case "error":
            BuggerLog.warning("BitableEventService server error: \(data)")

        case "connected", "heartbeat":
            break  // no-op

        default:
            break  // unknown event, ignore
        }
    }
}

4.2 Modify: `Sources/Models/AppConfig.swift`

Add the feishuAppBaseURL field:

// Add to the AppConfig struct:
var feishuAppBaseURL: String  // e.g. "http://192.168.1.100:8000"

Also update isConfigured to include this if SSE is the primary refresh mechanism (optional — polling still works without it):

// No change needed — isConfigured only requires appToken + tableId.
// SSE is an enhancement, not a requirement.

4.3 Modify: `Sources/AppDelegate.swift`

Start the SSE connection on launch:

// In applicationDidFinishLaunching, after PollerService startup (around line 32):
Task {
    await PollerService.shared.startIfConfigured()
    BitableEventService.shared.connect()  // SSE is non-blocking
}

4.4 Modify: `Sources/Views/Settings/SettingsView.swift`

Add the feishu-app URL field to the settings UI. The user configures the feishu-app server address (e.g. http://192.168.1.100:8000 for a shared deployment, or http://localhost:8000 for local Docker).

5. How Assignee Filtering Works (End-to-End)

Given three developers — Alice, Bob, Carol — each running Bugger configured with their own name:

feishu-app cache (file_token=X, assignee_field="负责人"):

  rec_001 → "Alice"
  rec_002 → "Bob"
  rec_003 → "Carol"
  rec_004 → "Bob"

Bugger SSE connections active:
  file_token=X, assignee_name=Alice   (Alice's Bugger)
  file_token=X, assignee_name=Bob     (Bob's Bugger)
  # Carol's Bugger is closed today

Scenario 1: Tester changes priority of rec_001

Event: record rec_001 changed
feishu-app: fetch rec_001 → assignee="Alice"
Cache: cached="Alice", current="Alice" → same → SKIP
Notifications: None. Zero Bugger fetches.

Scenario 2: PM reassigns rec_001 from Alice → Bob

Event: record rec_001 changed
feishu-app: fetch rec_001 → assignee="Bob"
Cache: cached="Alice", current="Bob" → changed!
  → Update cache: rec_001 → "Bob"
  → Notify Bob's SSE connection
  → Notify Alice's SSE connection
Alice's Bugger: fetchNow() → Alice's bugs list → rec_001 gone ✓
Bob's Bugger: fetchNow() → Bob's bugs list → rec_001 appears ✓

Scenario 3: New bug created, assigned to Carol

Event: record rec_005 created
feishu-app: fetch rec_005 → assignee="Carol"
Cache: no entry for rec_005 → new record
  → Update cache: rec_005 → "Carol"
  → Notify Carol's SSE connection... but Carol's Bugger is closed!
  → No notification delivered.

Next morning (9:00): Carol opens Bugger → PollerService.fetchNow() runs immediately
  → Carol sees the bug.
  OR: daily reconciliation detects rec_005 as new → but Carol still offline → no-op.

Scenario 4: Record deleted

Event: record rec_003 deleted
feishu-app: fetch rec_003 → 404
Cache: rec_003 → "Carol" (cached)
  → Remove from cache
  → Notify "Carol" connections
Carol's Bugger (when online): fetchNow() → rec_003 gone ✓

6. Config Changes Summary

feishu-app `config.py` — New Setting

Field	Type	Default	Description
`bitable_change_notification_enabled`	`bool`	`false`	Master feature flag. All resources are gated behind this.

feishu-app `.env` — New Entry

# Enable Bitable change notification (SSE push to Bugger)
BITABLE_CHANGE_NOTIFICATION_ENABLED=false

Bugger `AppConfig` — New Field

Field	Type	Default	Description
`feishuAppBaseURL`	`String`	`""`	feishu-app server URL, e.g. `http://192.168.1.100:8000`

When feishuAppBaseURL is empty, Bugger's SSE client treats it as disabled and never attempts a connection — no HTTP requests leave the Mac.

Bugger Settings UI — New Input

Add a text field in Settings labeled "Feishu App URL" for the feishuAppBaseURL config value.

7. Testing

7.1 Test feishu-app in Isolation

# 0. Enable the feature flag
export BITABLE_CHANGE_NOTIFICATION_ENABLED=true

# 1. Start feishu-app
cd ../../aptsell/feishu-app/feishu-service-python
python run_feishu_service.py

# 2. Verify the feature is loaded
#    Look for log line: "✅ Bitable 事件服务已初始化"
#    Look for log line: "已注册 bitable_record_changed 处理器"

# 3. Test SSE endpoint with curl
curl -N "http://localhost:8000/api/v1/bitable/events?file_token=YOUR_TOKEN&assignee_field=负责人&assignee_name=张三&table_id=YOUR_TABLE_ID"

# Expected: SSE stream with initial "connected" event
# event: connected
# data: {"status":"ok"}

# 4. Make a change in the Bitable (change an assignee)
#    → Watch the curl SSE stream for a "change" event

# 5. Verify cache warm logged correctly
#    Look for: "Cache warmed for ... (field=负责人): N records"

7.2 Test with Bugger

Start feishu-app
Configure feishuAppBaseURL in Bugger Settings to http://localhost:8000
Configure a Bitable with the correct appToken, tableId, and assignee field mapping
Restart Bugger → check logs for BitableEventService: connecting to ...
Edit a record's assignee in Feishu → Bugger should refresh within seconds
Check Bugger logs for "Bitable change push received, triggering fetchNow()"

7.3 Test Error Cases

Scenario	Expected Behavior
Feature flag disabled (default)	feishu-app starts normally. No Bitable imports, no Redis cache keys, no SSE endpoint, no WebSocket handler. Zero overhead. Bugger uses polling as always.
feishu-app not running	Bugger SSE fails to connect, retries every 30s. `PollerService` covers the gap.
App not added to Bitable	`_subscribe_to_file` logs error code 1069603. Admin must add the app as document manager.
Bugger closed	SSE connection dropped. feishu-app cleans up connection; daily reconciliation catches missed events.
WebSocket disconnect	feishu-app SDK auto-reconnects. Cache stays warm. Daily reconciliation fills any gaps.
Duplicate events	Redis-based dedup via `event_id` prevents double-processing.
SSE connection dies mid-stream	`URLSessionDataDelegate.didCompleteWithError` triggers auto-reconnect after 30s.
SSE endpoint called when flag disabled	Returns 404 (route not registered).

8. Files Changed Summary

feishu-app (separate repo: `../../aptsell/feishu-app/feishu-service-python/`)

File	Action	LOC
`src/app/services/bitable_event_service.py`	New	~280
`src/app/api/endpoints/bitable_subscription.py`	New	~80
`src/app/config.py`	Modify	+3
`run_feishu_service.py`	Modify	+45
`src/app/main.py`	Modify	+12
`.env` (or `.env.example`)	Modify	+2
Total (Python)		~420 LOC

Bugger (this repo)

File	Action	LOC
`Sources/Services/BitableEventService.swift`	New	~100
`Sources/Models/AppConfig.swift`	Modify	+2
`Sources/AppDelegate.swift`	Modify	+2
`Sources/Views/Settings/SettingsView.swift`	Modify	+15
Total (Swift)		~120 LOC

9. Deployment Considerations

9.1 feishu-app Network Reachability

feishu-app is a Docker service. Bugger must be able to reach it over the network:

Shared server deployment (current): Bugger users configure feishuAppBaseURL to the server address (e.g. http://10.0.1.50:8000). SSE is outbound from Bugger → works through NAT.
Local Docker per user: each developer runs feishu-app locally, feishuAppBaseURL = http://localhost:8000.

Both work because Bugger initiates the connection.

9.2 SSE Connection Count

feishu-app holds one SSE connection per active Bugger instance. For a team of 20, that's up to 20 idle TCP connections — negligible for any modern server.

9.3 Nginx / Reverse Proxy

If feishu-app sits behind nginx, add these directives to prevent SSE buffering:

location /api/v1/bitable/events {
    proxy_pass http://feishu-app:8000;
    proxy_http_version 1.1;
    proxy_set_header Connection "";
    proxy_buffering off;
    proxy_cache off;
    proxy_read_timeout 86400s;
}

9.4 Security

The SSE endpoint accepts query parameters with no authentication (v1). For production:

Add a shared API token in Settings → validate via Authorization header
Or restrict the feishu-app port to the VPN/office network only

10. Limitations & Future Work

Limitation	Mitigation
Formula field changes don't trigger events	`PollerService` periodic poll catches these
Bulk import may produce many events	Dedup by `event_id`; record fetches are idempotent
App must be added per Bitable	One-time setup; feishu-app logs clear error on failure
feishu-app downtime	`PollerService` is the universal fallback
Assignee field must be configured consistently	Bugger's existing `AppConfig.fieldMappings.assignee` provides this
Multi-field assignee detection	Cache is keyed per `{file_token, assignee_field}`; multiple field names can be tracked
Cache cold after feishu-app restart	SSE connections trigger cache warm on connect; reconciliation fills gaps

48 KiB Raw Blame History