mirror of
https://github.com/multica-ai/multica.git
synced 2026-08-10 06:49:21 +02:00
* fix: ingest feishu media as chat attachments * fix: ingest feishu post embedded media * fix(lark): make inbound media retries safe * fix lark media resource limit * fix(lark): move inbound media off ack path * fix(channel): make inbound media runs durable * fix(channel): close enqueue-vs-append race on media deferral EnqueueChatTask read the session-wide media deadline in one statement and sealed the input batch in a later one. Under READ COMMITTED a media message committing between the two got sealed into a task the deadline read had already decided was 'queued', so the daemon could claim it before its attachment bound — the agent received the bare placeholder, and the later media-ready promotion was a no-op against a non-deferred task. After the seal, re-derive the deferral from the sealed batch itself in the same transaction (DeferChatTaskForSealedPendingMedia): if any sealed message still carries an unexpired media marker, flip the task to deferred with fire_at aligned to the latest marker. The existing post-commit promote fence already covers the opposite direction (marker cleared mid-transaction). Adds a deterministic regression test that injects the media append between the deadline read and the seal via a wrapped pgx.Tx. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): keep committed chat task out of enqueue error path The post-commit media-ready fence returned its error from EnqueueChatTask even though the deferred task was already durably committed. The router flush treats any enqueue error as "no task exists": it clears the typing indicator and logs an enqueue failure while the run still happens at its fire_at deadline. Log the fence failure instead — the claim-path deferred promoter re-queues the task regardless. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): cap global media resolution concurrency Media jobs were serialized per session but unbounded across sessions: a burst could open arbitrarily many concurrent 45s Lark downloads, and each unknown-length upload may buffer up to the 100 MiB resource cap in memory. Gate resolveAndBindMedia behind a global slot semaphore (default 8, RouterConfig.MediaConcurrency). Per-session ordering is unchanged; on shutdown a job cancelled while waiting for a slot proceeds straight to the bounded DB finalize so marker clearing stays prompt. Also document that the per-message media budget spans queue/slot waits (it must match the persisted fire_at) and why timed-out uploads cannot leak unbounded orphans. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(chat): keep channel-sealed user messages on task cancel Sealing the channel input batch stamps task_id onto channel user messages, which exposed them to the cancel draft-restore path: an empty-transcript cancel would DeleteUserChatMessageByTask the sealed Feishu/Slack messages and detach their attachments. Those messages are the durable record of what the platform sender wrote — the sender has no Multica composer to restore a draft into. Gate the restore-delete on ChatSessionHasChannelBinding in both the synchronous finalize and the deferred finalize (the latter covers markers left by an older replica during a rolling deploy); a bound session now settles as "Stopped." instead. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): skip the media pipeline for messages without media Every inbound message on a Media-enabled platform persisted a 45s media deadline and queued a resolution job, so a plain text message could wait behind the global media semaphore (its task deferred while other sessions download 100 MiB videos) and a crash between append and clear delayed a pure-text run to the full 45s fallback. Add MediaResolver.HasMedia — a pure in-memory probe the Router calls on the ACK path — and only persist the deadline / enqueue the job when the message actually references platform media. The Feishu resolver decodes the already-received payload and reports standalone image or video keys and post-embedded img/media spans. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(chat): gate cancel restore on immutable channel provenance The previous guard keyed the cancel restore-delete off ChatSessionHasChannelBinding, but a binding only proves routing exists right now: archiving a session and rebinding an installation both delete the binding while preserving chat history, so a still-cancellable sealed task could again restore-delete the original inbound messages. Persist provenance on the message instead: migration 203 adds chat_message.channel_ingested, stamped inside the channel append transaction and never mutated, and both cancel finalize paths now gate on TaskHasChannelIngestedMessages over the task's sealed batch. The binding-existence query is removed. Regression tests cover ingest -> archive/unbind -> cancel for a queued and a started task. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): reclaim media uploads that never gain an attachment row Deadline expiry dropped already-resolved refs and a BindMedia failure was log-only, leaving uploaded objects with no attachment row and no reclaim path — the dedup mark commits with the message before media runs, so a redelivery is dropped as a duplicate and never re-resolves (and thus never overwrites) those keys, and workspace/session deletion only enumerates the attachment table. Add MediaResolver.DiscardMedia — a best-effort delete by StorageKey — and call it from both failure paths in resolveAndBindMedia. The Feishu resolver forwards to the storage backend's Delete. Tests cover a partial upload discarded at the deadline, discard on bind failure, and key-level deletion in the resolver. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * docs(server): refresh comments stale after detached media ingestion Channel tasks now seal a self-owned input batch, media ingestion is no longer out of scope for the flattener, and MediaRefs are filled by the detached resolver after append rather than by feishuChannel pre-engine. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(chat): stop keying channel empty-completion silence off chat_input_task_id Sealing gave channel tasks a self input-owner, which broke writeChatCompletionOutcome's discriminator: it treated any owned task as direct, so an empty channel completion wrote the no_response fallback row and the outbound patcher — which forwards any non-empty chat:done content verbatim — pushed the English fallback body to Feishu/Slack, violating the MUL-4351 contract. Silence is now decided by the immutable channel_ingested provenance of the task's input batch, looked up by the batch OWNER id (chat_input_task_id): auto-retry clones inherit the owner while their sealed messages stay tagged with the parent's id, so keying off the task's own id would misread a channel retry as direct. The cancel-path provenance gates switch to the same owner key via chatInputOwnerID. chat_input_task_id is back to meaning only "input batch owner". Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * chore(migrations): renumber to 203/204 after upstream took 202 Upstream main merged 202_runtime_profile_add_qwen while this branch held 202/203, tripping TestMigrationNumericPrefixesStayUniqueAfterLegacySet on the CI merge tree. channel_media_pending becomes 203 and channel_ingested becomes 204; no content changes. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channels): gate outbound delivery on channel provenance, not owner Merging main brought #5645 (keep direct chat replies in Multica), whose outbound gate assumed channel tasks leave chat_input_task_id NULL. Sealed channel tasks own an input batch too, so on the merge tree every channel reply and failure notice was classified as direct and silently dropped — agents stopped replying in Feishu/Slack. Both outbound gates now call engine.TaskInputIsChannelIngested: a NULL owner keeps #5645's deliver-by-default for pre-sealing tasks, an owned batch delivers only when it carries the immutable channel_ingested stamp (keyed by the owner id, so auto-retry clones inherit the verdict). Direct replies stay in Multica; sealed channel replies reach the platform. Tests cover both directions on both platforms. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): discard media orphans on a fresh context, after finalize DiscardMedia shared finalizeCtx with BindMedia, so a bind that failed because the finalize deadline expired handed the storage deletes an already-dead context — the compensation silently no-opped and the orphans leaked anyway. The deadline path also ran S3 deletes before the marker clear, eating the same 5s budget the user-facing bind/promotion needed. Collect the refs from both failure paths, run bind + promotion on the finalize budget first, then delete on a fresh discard context. The bind-failure test now pins that discard receives a live context. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): compensate result-uncertain media uploads and commits The compensation protocol treated "the call returned an error" as "the side effect did not happen", which is wrong in both directions across the result-uncertain windows: - An upload error can follow a server-side write (lost response, deadline mid-write). The attempted key never reached the router, so nothing could reclaim it and dedup guarantees no re-resolve. The resolver now idempotently deletes the deterministic key on a fresh budget right at the failure site. - A commit error is not a rollback guarantee: a lost ack can report failure after Postgres durably committed the attachment rows, and the router's discard would then delete objects those rows reference. BindMediaRefs now converges the ambiguity on a fresh budget — any of the batch's URLs present proves the atomic commit landed (bind reports success); none proves the rollback (discard stays safe); a failed verification returns ErrMediaBindResultUnknown and the router keeps the uploads, preferring a rare orphan over a broken attachment. Fault-injection coverage: an upload error deletes the attempted key; a lost-ack commit keeps the bound attachment and reports success; a verified rollback stays a discardable error; the router keeps uploads on the unknown-outcome sentinel. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * chore(migrations): renumber to 207/208 after upstream took 203-206 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * docs(channel): note DiscardMedia self-invocation and the unknown-outcome skip Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * chore(migrations): renumber to 212/213 after upstream took 207-211 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * feat(channel): replace inline media compensation with an intent ledger and reconciler Inline best-effort compensation cannot answer "did my side effect happen?" at the moment it needs the answer — the DELETE/PUT reordering and the empty-read-vs-in-flight-COMMIT gaps were both instances of the same two-system atomicity problem. Persist the intent instead and let an asynchronous reconciler settle it: - channel_media_pending_object (migration 214; claim index 215 as its own single-statement CONCURRENTLY migration): a state machine row ('pending' -> 'deleting') with lease, attempt, and backoff columns. - The resolver upserts the row BEFORE each PUT, state-guarded so a key the reconciler owns is never resurrected (the resource is skipped). ObjectURL is a pure function of configuration, so the row carries the attachment URL pre-upload. - BindMediaRefs deletes the batch's rows INSIDE the attachment-insert transaction: commit landed <=> intents gone, atomically, so an ambiguous COMMIT never needs adjudication. A key already claimed to 'deleting' is skipped (placeholder stays). - Nothing is ever deleted inline. The reconciler — an independent worker so storage latency cannot starve other sweepers — claims due rows ('pending' past the settle delay, or expired leases) under a fresh lease, checks for a durable attachment reference only AFTER the claim (race-free: bind can no longer succeed on the key), deletes unreferenced objects outside any transaction, and backs off failed deletes with attempt-based retry. Crash windows converge for free. - The settle delay is a fixed constant carrying NO correctness weight; invariant tests pin it at >=10x every pipeline budget. Metrics cover deletes, referenced clears, delete failures, and ledger backlog. Removed: MediaResolver.DiscardMedia, ErrMediaBindResultUnknown, the post-commit verification, and both router discard branches. Tests: intent-before-upload ordering; upload error leaves the row and deletes nothing; bind-wins vs reconciler-wins on the same key; lost-ack and rolled-back commit injections (intent cleared iff the attachment landed); reconciler three-state settle; expired-lease reclaim; delete failure backoff and retry; settle invariants. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): never build or sweep the media reconciler without storage store is nil when S3 is unconfigured AND the local upload dir fails to initialize, but the reconciler was constructed unconditionally and main only gates the goroutine on the reconciler pointer — the first unreferenced ledger row (rows can pre-exist from a boot where storage worked) would nil-pointer panic a bare goroutine and take down the process. Construct the reconciler only when a storage backend exists, and guard RunOnce defensively: with no deleter it skips the sweep without claiming, so rows are not stranded in 'deleting' until lease expiry. Test covers the pre-existing-row + missing-storage boot. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * chore(migrations): renumber to 213-216 after upstream took 212 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * refactor(channel): remove the dead pre-resolved MediaRefs ingress path lark.InboundMessage.MediaRefs and the resolver's early-returns for pre-populated refs were vestiges of the pre-detached synchronous design — no producer fills them before the router anymore. Worse, the intent ledger made the path actively misleading: refs arriving without ledger rows would be silently skipped at bind (with a log blaming the reconciler), contradicting the field's "already persisted" contract. Delete the field, its channelMessageFromLark mapping, and both early-returns; channel.InboundMessage.MediaRefs is now documented as what it actually is — ResolveMedia's output channel, always empty on ingress, attachable only through a claimed ledger intent. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): enforce workspace tenancy on every ledger query The intent-ledger upsert's conflict branch guarded only on state, so a cross-workspace storage_key collision could rewrite the row's workspace/message/url ownership; release and delete keyed on (storage_key, lease_token) alone. The derived key embeds the workspace UUID so none of this is reachable today — but tenancy must be enforced by the workspace column in every query, never derived from the key string (MUL-3515 rule, restated in this PR's review). The upsert now updates only within the same workspace (a cross-tenant conflict updates nothing, returns no row, and the resolver skips the upload — the fail-safe direction), and release/delete take (workspace_id, storage_key, lease_token). Tests pin that a foreign workspace can neither steal, release, nor delete a row. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(migrations): build the ledger primary key via a concurrent index storage_key TEXT PRIMARY KEY created its unique index implicitly at CREATE TABLE, against the repo convention that every migration index — including a new table's unique index — is built CONCURRENTLY in its own single-statement migration (the exact three-step pattern client_usage_daily shipped in 207-209). The table now declares storage_key NOT NULL, 216 builds the unique index concurrently, and 217 attaches the primary key USING INDEX; the claim index moves to 218. ON CONFLICT (storage_key) still resolves against the constraint, and the full down/up round-trip is verified. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): bound each reconciler object delete with its own timeout DeleteObject ran on the worker-lifetime context and the SDK's default HTTP client has no overall request timeout, so one black-holed connection would wedge the sequential sweep loop — and with it every later batch and the backlog gauge — forever; a single-replica deployment has no other worker to reclaim the lease. Each delete now gets a 30s timeout (well under the 2min lease), and a timed-out delete takes the existing release/backoff path. Covered by a blocking-deleter test with an injectable timeout. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): anchor the media deadline to the DB clock and bound queue waits by it Two deadline gaps from review: - The persisted marker was an application-clock timestamp compared against SQL now() everywhere it is read, so a skewed app node could shrink the fallback window and hand the agent a placeholder before the resolver's local budget ended. The append transaction now anchors a relative budget (MediaPendingSeconds) with now() + make_interval, writer and readers sharing one clock; the local resolve budget stays monotonic app-side. A DB test pins that the remaining budget measured by the DB clock equals the requested one. - enqueueMedia's waits (per-session order, global slot) only watched shutdown, so in a burst an already-expired job kept its goroutine and payload until it reached the front. Both waits now also watch the message's deadline; on expiry the job skips the resolver entirely and runs only the empty finalize (marker clear + promotion), which also unblocks the session's later messages. Covered by a queued-expiry test that finalizes while the only slot is deterministically held. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * chore(migrations): renumber to 216-221 after upstream took 213-215 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): start the local media budget before the append transaction The DB anchors the durable fallback at insert-time now(), but the local monotonic budget started only after AppendMessage returned — so the resolver outlived the fallback by the append/commit latency, a window where the deferred task is already claimable while the resolver still runs and the agent reads a placeholder that binds moments later. Capture the local deadline before calling AppendMessage, restoring the ordering local-gives-up <= durable-fallback-fires. A slow-append test pins that the resolver's context deadline is measured from the pre-append instant. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * chore(migrations): renumber to 224-229 after upstream took 216-223 Verified against the merged tree: the numeric-prefix uniqueness test passes and the full migration set applies cleanly from scratch. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): heartbeat the reconciler lease per row One claim covers up to 50 rows under a single 2-minute lease, but the batch is processed sequentially and each delete may run its full 30s timeout — a few stalled deletes could outlive the lease mid-batch, letting another replica reclaim the tail: duplicate concurrent deletes, inflated attempt/backoff on rows whose owner was alive, and skewed metrics. The lease is now renewed before EACH row's settle work, so it only ever needs to cover one row's worst case (invariant-tested: lease >= 2x the per-delete timeout). A renewal that matches no row means another worker reclaimed it after a genuine expiry — the row is skipped, leaving the new owner's state untouched. Test simulates a mid-batch reclaim and pins the skip. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): dedup post media resources and make local writes atomic A rich post may reference the same image_key/file_key in several spans. The object key derives from (message, type, key), so duplicates uploaded to the SAME key twice: LocalStorage.UploadStream truncated the destination up front and removed it outright on a copy error, so a second failing attempt destroyed the object the first success had produced — leaving an attachment row pointing at nothing. A second succeeding attempt instead produced two attachment rows for one object. Collapse duplicate spans by (fetch type, platform key) before the upload loop, and write local uploads through a temp file renamed into place so a failed write can only discard its own temp file. Tests cover a duplicated span uploading once and a failed re-upload leaving the previous object intact. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): fence late-materializing PUTs with a tombstone schedule A DELETE cannot be ordered against a PUT the client already abandoned: the store may materialize the object after the delete completes. The reconciler cleared the ledger row right after deleting, so such an object had no row and nothing to reclaim it — which made the settle delay the de-facto correctness barrier for the PUT/DELETE race, exactly what the design says it must not be. The row is now kept as a tombstone ('tombstoned' state, migration 226's CHECK) and re-deleted on a widening schedule (15m, 1h, 6h, 24h, the pass index carried in last_error), so a late materialization is reclaimed by a later pass; only after the schedule is exhausted is the row dropped. Claim, heartbeat, lease, and tenancy predicates are unchanged — a tombstone is claimed exactly like any other due row. A separate gauge reports tombstones so they cannot be mistaken for a backlog of objects awaiting reclaim, and the header comment now states precisely what state fences (bind/commit) versus what the schedule fences (late PUTs). Tests: the reviewer's interleaving — DELETE completes, the abandoned PUT materializes right after, and the object is gone by the end of the schedule — plus a full schedule walk asserting the object is counted once and the row clears at the end. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): tombstones must re-delete, not re-ask the reference question A tombstone revisit ran the same reference check as a first settle, so an attachment carrying the same URL — a re-ingested copy of the object — sent the row down the "referenced, keep it" branch: the object was kept and the row cleared, abandoning the re-delete schedule that fences the ORIGINAL object against an abandoned PUT. A tombstone has already been judged unreferenced and deleted; it exists only to re-delete whatever materializes later, so it now goes straight to the delete + schedule tail (extracted as settleDeletedObject, shared with the first settle). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): keep the tombstone schedule position in its own column The re-delete pass index was encoded into last_error, which the failure path also writes: one failed re-delete erased the position and restarted the walk. A store failing intermittently could therefore keep a tombstone alive indefinitely — every recovery would resume at pass 1 and the row would never reach the end of the schedule to be dropped. tombstone_pass is now its own column (the table is introduced in this PR, so migration 226 carries it), advanced only by a successful delete, and the tombstone write clears the now-stale last_error. Test walks the schedule across a failed re-delete and asserts it resumes rather than restarts, and that the row still terminates. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(lark): derive media object keys per chat message The object key was derived from the platform message alone, so a second ingest of the same Feishu message reused the first ingest's ledger row. That row can be a tombstone (up to ~31h while the re-delete schedule runs), and the intent upsert refuses anything that has left 'pending', so the second ingest skipped the upload and silently produced a placeholder with no attachment. A re-ingest is reachable: the inbound dedup claim is reclaimable once 60s stale and the dedup row is only vacuumed after 24h. Keying on the chat message the object will attach to keeps the two ingests independent, and nothing leaks: each one's objects are covered by its own ledger row. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * refactor(storage): route both local upload paths through one atomic write UploadStream wrote through a temp file and renamed into place, but the buffered Upload path still truncated the destination up front — the destructive shape the stream path exists to avoid, one caller away from coming back. Both now share writeAtomic, which also restores the 0644 the direct write used (CreateTemp makes files 0600). Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * chore(channel): gofmt the media-pending append fields Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(storage): keep the local upload chmod best-effort The rename-into-place rewrite made a failed chmod fail the whole upload. CreateTemp's 0600 has to be widened to the 0644 the direct write used, but an upload dir on a mount that ignores chmod (SMB/NFS/FUSE) accepted the old direct write fine — turning those deployments' uploads into hard errors would be a regression for a cosmetic property. Log and continue. Tests pin 0644 on both upload paths, and that a failed buffered upload leaves no temp litter and no damage to a previous object. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(channel): never re-delete an object an attachment references The tombstone pass skipped the reference check and deleted unconditionally, so a durable attachment carrying that URL lost the only object it can read — the dangling attachment the intent ledger exists to prevent, and the opposite of the posture every other path here takes ("a reclaimable orphan beats a broken attachment"). The check now runs on every pass. A positive result on a tombstone is unreachable by design — keys are per (chat message, resource) and a bind cannot attach a key that has left 'pending' — so reaching it means an invariant broke: keep the object, clear the row, log it, and count it on a dedicated reconciler_tombstone_referenced_total counter. The test's contract is flipped to assert the referenced object survives. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(storage): make the local staging file reclaimable after a crash os.CreateTemp's random suffix meant a crash between the staging write and the rename left a file nothing could name: the ledger records only the final storage key, and DeleteObject removed only the object and its sidecar. Each leftover can approach the 100 MiB resource cap and they accumulate without bound. The staging path is now derived from the object key, so DeleteObject removes it alongside the object — which makes the media reconciler reclaim it too, since the intent row is written before the upload. Opening it 0644 directly also drops the chmod the previous commit had to make best-effort. Both read paths refuse the staging name (keys come from the request URL, and a half-written body should not be readable); a user-supplied ".tmp" extension is unaffected, since object keys are generated and never dot-prefixed. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * chore(channel): renumber the media migrations after merging main main took 224 (agent_task_session_rollout_missing), so the ledger group moves to 225-230 and the cross-references inside the table migration follow. main's CompleteTask also grew a sessionRolloutMissing parameter; the three call sites this PR added to chat_input_ownership_test.go pass false. Verified the way the numbering is meant to be verified: full migration set applied from scratch on the merged tree, and the whole server suite run against that database. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(channel): let Postgres compute every reconciler deadline The reconciler built settle cutoffs, lease expiry, backoff and re-delete times from the process clock and compared them against the database's now(). A replica whose clock had drifted would therefore settle rows whose upload was still in flight (the object is deleted and the bind then refuses to attach — media silently lost), hand out leases that are born expired (rows churn between workers, attempt/backoff inflate), or compress the tombstone schedule that fences a late-materializing PUT. The four settle queries now take durations and derive their timestamps from now(), so every replica reads one clock. The parameter types are the guard: an app-side timestamp can no longer be passed. Test asserts the persisted lease, backoff and re-delete deadlines all track the database's now(). The generated code also picks up main's new agent_task_queue column in the two RETURNING task.* queries this PR adds. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * chore(lark): drop unrelated gofmt-only churn from this PR Six files carried whitespace/comment-reformatting with no functional change, unrelated to the inbound media pipeline. Reverting them to the base revision keeps the diff focused on the feature (75 -> 69 files): server/internal/service/empty_claim_cache.go server/internal/integrations/lark/markdown_detect.go server/internal/integrations/lark/ws_chunk_assembler.go server/internal/integrations/lark/ws_chunk_assembler_test.go server/internal/integrations/lark/ws_frame_test.go server/internal/integrations/lark/registration_test.go Verified: `git diff -w` against these files was already empty, so no behavior is affected. go vet clean; tests covering these files pass. Co-authored-by: multica-agent <github@multica.ai> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com> Co-authored-by: Bohan-J <bohan@devv.ai> Co-authored-by: multica-agent <github@multica.ai>
1259 lines
46 KiB
Go
1259 lines
46 KiB
Go
package lark
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"log/slog"
|
||
"mime"
|
||
"net/http"
|
||
"net/url"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
)
|
||
|
||
// Real Lark/飞书 Open Platform HTTP APIClient.
|
||
//
|
||
// Scope: tenant_access_token acquisition + caching, IM v1 interactive-
|
||
// card send / patch, the dedicated binding-prompt outbound, AND the
|
||
// install-time Bot identity lookup (/open-apis/bot/v3/info) consumed
|
||
// by RegistrationService right after a successful device-flow grant.
|
||
// The PersonalAgent registration protocol itself is a separate client
|
||
// (RegistrationClient) because it speaks to a different host
|
||
// (accounts.feishu.cn) with a different auth model (no
|
||
// tenant_access_token — the response IS the credentials).
|
||
//
|
||
// Per-installation credentials flow in on each call via
|
||
// InstallationCredentials; the client never reads lark_installation
|
||
// directly. tenant_access_token is cached in-process keyed by app_id,
|
||
// honoring Lark's `expire` field minus a safety margin so callers
|
||
// never present a token that's about to lapse mid-flight.
|
||
|
||
// DefaultResourceDownloadTimeout is the default cap on one message-resource
|
||
// download. Exported so the channel-media settle invariant test can assert
|
||
// the reconciler's settle delay dwarfs every pipeline budget.
|
||
const DefaultResourceDownloadTimeout = 45 * time.Second
|
||
|
||
const (
|
||
// defaultLarkBaseURL is the mainland 飞书 open-platform host. It is the
|
||
// fallback host for an installation whose region is feishu (or unset);
|
||
// Region.OpenPlatformBaseURL maps region=lark to open.larksuite.com.
|
||
// Operators do NOT set MULTICA_LARK_HTTP_BASE_URL to pick a cloud
|
||
// anymore — the per-installation region does that automatically. The
|
||
// env var remains only as a deployment-wide override (proxy / mock /
|
||
// single-cloud staging); tests substitute an httptest.Server URL.
|
||
defaultLarkBaseURL = "https://open.feishu.cn"
|
||
|
||
// tokenSafetyMargin is subtracted from Lark's `expire` so we
|
||
// refresh before a token actually lapses. 60s comfortably exceeds
|
||
// any in-flight HTTP timeout we set below.
|
||
tokenSafetyMargin = 60 * time.Second
|
||
|
||
// defaultRequestTimeout is the per-call HTTP timeout. Lark's API
|
||
// is normally well under 1s; we leave headroom for cross-region
|
||
// latency from a self-hosted Multica deployment to feishu.cn.
|
||
defaultRequestTimeout = 10 * time.Second
|
||
|
||
// defaultResourceDownloadTimeout is intentionally longer than normal
|
||
// OpenAPI calls because message videos are binary transfers, not JSON
|
||
// RPCs. Keep it below the inbound dedup stale-claim window (60s), so a
|
||
// slow download does not invite a second replica to reclaim the same
|
||
// message before this one can append and mark it processed.
|
||
defaultResourceDownloadTimeout = DefaultResourceDownloadTimeout
|
||
|
||
// Feishu caps message resources at 100 MiB. Keep the local transport guard
|
||
// aligned with that contract; detached media processing keeps large
|
||
// transfers off the connector ACK path.
|
||
maxMessageResourceBytes = 100 << 20
|
||
|
||
// Lark's "invalid tenant_access_token" / "tenant_access_token
|
||
// expired" error codes. When we see either, drop the cached token
|
||
// so the next call refreshes from /tenant_access_token/internal.
|
||
// 99991663 = expired, 99991664 = invalid. Documented at:
|
||
// open.feishu.cn/document/server-docs/api-call-guide/server-error-codes.
|
||
codeTokenExpired = 99991663
|
||
codeTokenInvalid = 99991664
|
||
)
|
||
|
||
// HTTPClientConfig configures the production Lark HTTP APIClient.
|
||
type HTTPClientConfig struct {
|
||
// BaseURL is an optional deployment-wide override for the Lark
|
||
// open-platform root, e.g. "https://open.feishu.cn" or
|
||
// "https://open.larksuite.com". When set it forces every call —
|
||
// regardless of the installation's region — to that host; tests set
|
||
// it to an httptest.Server URL. When EMPTY (the production default),
|
||
// each call resolves its host from InstallationCredentials.Region so
|
||
// a single deployment serves both Feishu and Lark. Trailing "/" is
|
||
// stripped.
|
||
BaseURL string
|
||
|
||
// HTTPClient is the transport used for every outbound call. Tests
|
||
// substitute an *http.Client whose Transport routes to an
|
||
// httptest.Server. Empty defaults to a fresh http.Client with
|
||
// defaultRequestTimeout.
|
||
HTTPClient *http.Client
|
||
|
||
// ResourceHTTPClient is used only for message resource downloads. It
|
||
// deliberately does not share HTTPClient's shorter timeout: image/video
|
||
// resource transfers are bounded by ResourceDownloadTimeout instead.
|
||
ResourceHTTPClient *http.Client
|
||
|
||
// ResourceDownloadTimeout caps a single message resource download. Zero
|
||
// defaults to defaultResourceDownloadTimeout.
|
||
ResourceDownloadTimeout time.Duration
|
||
|
||
// Now is overridable for deterministic token-expiry tests.
|
||
Now func() time.Time
|
||
|
||
// Logger receives warnings about Lark error codes. Nil uses
|
||
// slog.Default().
|
||
Logger *slog.Logger
|
||
}
|
||
|
||
func (c HTTPClientConfig) withDefaults() HTTPClientConfig {
|
||
// BaseURL is intentionally NOT defaulted to defaultLarkBaseURL here.
|
||
// An empty BaseURL means "no deployment-wide override" — each call
|
||
// then resolves its host from InstallationCredentials.Region (see
|
||
// resolveBaseURL), so one client serves both Feishu and Lark. A
|
||
// non-empty BaseURL (MULTICA_LARK_HTTP_BASE_URL, or an httptest URL
|
||
// in tests) forces every region to that host.
|
||
c.BaseURL = strings.TrimRight(c.BaseURL, "/")
|
||
if c.HTTPClient == nil {
|
||
c.HTTPClient = &http.Client{Timeout: defaultRequestTimeout}
|
||
}
|
||
if c.ResourceDownloadTimeout == 0 {
|
||
c.ResourceDownloadTimeout = defaultResourceDownloadTimeout
|
||
}
|
||
if c.ResourceHTTPClient == nil {
|
||
c.ResourceHTTPClient = &http.Client{Timeout: c.ResourceDownloadTimeout}
|
||
}
|
||
if c.Now == nil {
|
||
c.Now = time.Now
|
||
}
|
||
if c.Logger == nil {
|
||
c.Logger = slog.Default()
|
||
}
|
||
return c
|
||
}
|
||
|
||
// NewHTTPAPIClient constructs the real APIClient that speaks to Lark's
|
||
// open platform over HTTPS. Per-installation credentials flow in via
|
||
// each call's InstallationCredentials parameter; tokens are cached
|
||
// keyed by app_id so a single Multica server reuses Lark's
|
||
// tenant_access_token across calls to the same app.
|
||
func NewHTTPAPIClient(cfg HTTPClientConfig) APIClient {
|
||
cfg = cfg.withDefaults()
|
||
return &httpAPIClient{cfg: cfg, tokens: make(map[string]*cachedToken)}
|
||
}
|
||
|
||
type httpAPIClient struct {
|
||
cfg HTTPClientConfig
|
||
|
||
mu sync.Mutex
|
||
// tokens caches tenant_access_token keyed by app_id only — NOT by
|
||
// (app_id, region). This is safe because a Lark/飞书 app_id (the
|
||
// "cli_..." credential) is globally unique across both clouds and an
|
||
// app exists on exactly one of them, so an app_id never maps to two
|
||
// regions. The DB enforces the same assumption with UNIQUE(app_id) on
|
||
// lark_installation. If Lark ever reused an app_id across clouds, both
|
||
// this cache key and that constraint would need region added.
|
||
tokens map[string]*cachedToken
|
||
}
|
||
|
||
type cachedToken struct {
|
||
value string
|
||
expiresAt time.Time
|
||
}
|
||
|
||
// IsConfigured reports true: once this client exists at all, the
|
||
// outbound transport path (send / patch / binding prompt / bot info)
|
||
// is wired. The stub returns false because every call there errors
|
||
// with ErrAPIClientNotConfigured; the real client is the inverse
|
||
// contract.
|
||
func (c *httpAPIClient) IsConfigured() bool { return true }
|
||
|
||
// tenantAccessToken returns a usable tenant_access_token for the
|
||
// given installation, reusing a cached token while it is alive (minus
|
||
// safety margin) and otherwise fetching a fresh one from Lark.
|
||
//
|
||
// Concurrent callers serialize on the per-client mutex during the
|
||
// uncached path; the cached path takes the mutex only for the lookup
|
||
// and releases before doing any I/O. Steady-state contention is
|
||
// therefore one map-read under the lock, not a per-call HTTP round
|
||
// trip.
|
||
func (c *httpAPIClient) tenantAccessToken(ctx context.Context, creds InstallationCredentials) (string, error) {
|
||
if creds.AppID == "" {
|
||
return "", errors.New("lark http client: missing app_id")
|
||
}
|
||
if creds.AppSecret == "" {
|
||
return "", errors.New("lark http client: missing app_secret")
|
||
}
|
||
|
||
now := c.cfg.Now()
|
||
c.mu.Lock()
|
||
if t, ok := c.tokens[creds.AppID]; ok && t.expiresAt.After(now) {
|
||
val := t.value
|
||
c.mu.Unlock()
|
||
return val, nil
|
||
}
|
||
c.mu.Unlock()
|
||
|
||
// Self-built (internal) app endpoint. Marketplace / multi-tenant
|
||
// apps would use /tenant_access_token/v3 with a different body
|
||
// shape; PersonalAgent in this MVP is per-workspace self-built so
|
||
// we stay on /internal.
|
||
body := map[string]string{
|
||
"app_id": creds.AppID,
|
||
"app_secret": creds.AppSecret,
|
||
}
|
||
var resp struct {
|
||
Code int `json:"code"`
|
||
Msg string `json:"msg"`
|
||
TenantAccessToken string `json:"tenant_access_token"`
|
||
Expire int64 `json:"expire"`
|
||
}
|
||
if err := c.doJSON(ctx, c.resolveBaseURL(creds), http.MethodPost, "/open-apis/auth/v3/tenant_access_token/internal", "", body, &resp); err != nil {
|
||
return "", fmt.Errorf("lark http client: tenant_access_token: %w", err)
|
||
}
|
||
if resp.Code != 0 || resp.TenantAccessToken == "" {
|
||
return "", fmt.Errorf("lark http client: tenant_access_token: code=%d msg=%q", resp.Code, resp.Msg)
|
||
}
|
||
|
||
expire := time.Duration(resp.Expire) * time.Second
|
||
// Clamp to >= 2× safety margin so a misbehaving upstream that
|
||
// returns a sub-minute expire never makes us cache a token that
|
||
// is already past its safe window.
|
||
if expire < tokenSafetyMargin*2 {
|
||
expire = tokenSafetyMargin * 2
|
||
}
|
||
expiresAt := c.cfg.Now().Add(expire - tokenSafetyMargin)
|
||
|
||
c.mu.Lock()
|
||
c.tokens[creds.AppID] = &cachedToken{value: resp.TenantAccessToken, expiresAt: expiresAt}
|
||
c.mu.Unlock()
|
||
|
||
return resp.TenantAccessToken, nil
|
||
}
|
||
|
||
// resolveBaseURL picks the open-platform host for one call. An explicit
|
||
// cfg.BaseURL (MULTICA_LARK_HTTP_BASE_URL, or an httptest URL in tests)
|
||
// overrides every region and routes all traffic there. With no override,
|
||
// the host comes from the installation's region, so Feishu and Lark
|
||
// installations served by the same process each reach their own cloud.
|
||
func (c *httpAPIClient) resolveBaseURL(creds InstallationCredentials) string {
|
||
if c.cfg.BaseURL != "" {
|
||
return c.cfg.BaseURL
|
||
}
|
||
return creds.Region.OpenPlatformBaseURL()
|
||
}
|
||
|
||
// invalidateToken drops the cached token for an app_id. Called when
|
||
// Lark surfaces an expired / invalid token error code so the next
|
||
// call refreshes instead of looping on a stale entry.
|
||
func (c *httpAPIClient) invalidateToken(appID string) {
|
||
c.mu.Lock()
|
||
delete(c.tokens, appID)
|
||
c.mu.Unlock()
|
||
}
|
||
|
||
// outboundMessageRequest builds the (path, body) the three send methods
|
||
// share. When target.IsSet() the message is routed through Lark's reply
|
||
// endpoint (POST /im/v1/messages/{message_id}/reply) so it threads back
|
||
// into the originating 话题 — reply_in_thread carries the target's
|
||
// InThread flag (Lark also keeps the reply in-thread automatically when
|
||
// the parent message already belongs to a thread). Otherwise the message
|
||
// goes to the chat-level send endpoint keyed by receive_id=chat_id, the
|
||
// historical behavior. Body is map[string]any (not map[string]string)
|
||
// because reply_in_thread is a bool.
|
||
func outboundMessageRequest(chatID ChatID, msgType, content string, target ReplyTarget) (string, map[string]any) {
|
||
if target.IsSet() {
|
||
return "/open-apis/im/v1/messages/" + url.PathEscape(target.MessageID) + "/reply", map[string]any{
|
||
"msg_type": msgType,
|
||
"content": content,
|
||
"reply_in_thread": target.InThread,
|
||
}
|
||
}
|
||
q := url.Values{}
|
||
q.Set("receive_id_type", "chat_id")
|
||
return "/open-apis/im/v1/messages?" + q.Encode(), map[string]any{
|
||
"receive_id": string(chatID),
|
||
"msg_type": msgType,
|
||
"content": content,
|
||
}
|
||
}
|
||
|
||
// SendInteractiveCard posts a fresh interactive card into a chat and
|
||
// returns Lark's message_id so the Patcher can target subsequent
|
||
// patches at the same card.
|
||
func (c *httpAPIClient) SendInteractiveCard(ctx context.Context, p SendCardParams) (string, error) {
|
||
if p.ChatID == "" {
|
||
return "", errors.New("lark http client: missing chat_id")
|
||
}
|
||
if p.CardJSON == "" {
|
||
return "", errors.New("lark http client: missing card json")
|
||
}
|
||
token, err := c.tenantAccessToken(ctx, p.InstallationID)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
path, body := outboundMessageRequest(p.ChatID, "interactive", p.CardJSON, p.ReplyTarget)
|
||
var resp struct {
|
||
Code int `json:"code"`
|
||
Msg string `json:"msg"`
|
||
Data struct {
|
||
MessageID string `json:"message_id"`
|
||
} `json:"data"`
|
||
}
|
||
if err := c.doJSON(ctx, c.resolveBaseURL(p.InstallationID), http.MethodPost, path, token, body, &resp); err != nil {
|
||
return "", fmt.Errorf("lark http client: send interactive card: %w", err)
|
||
}
|
||
if resp.Code != 0 || resp.Data.MessageID == "" {
|
||
if isTokenError(resp.Code) {
|
||
c.invalidateToken(p.InstallationID.AppID)
|
||
}
|
||
return "", &APIError{Op: "send interactive card", Code: resp.Code, Msg: resp.Msg}
|
||
}
|
||
return resp.Data.MessageID, nil
|
||
}
|
||
|
||
// SendTextMessage posts a plain text IM message into a Lark chat.
|
||
// This is the Patcher's primary outbound for agent chat replies —
|
||
// using a normal text bubble instead of an interactive card makes
|
||
// free-form replies feel like a native Lark conversation. The
|
||
// content envelope Lark expects is a JSON-encoded `{"text": "..."}`
|
||
// blob; we encode it here so callers pass raw text.
|
||
func (c *httpAPIClient) SendTextMessage(ctx context.Context, p SendTextParams) (string, error) {
|
||
if p.ChatID == "" {
|
||
return "", errors.New("lark http client: missing chat_id")
|
||
}
|
||
if p.Text == "" {
|
||
return "", errors.New("lark http client: missing text")
|
||
}
|
||
token, err := c.tenantAccessToken(ctx, p.InstallationID)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
// Lark's `text` msg_type expects content = JSON-encoded {"text": "..."}.
|
||
// json.Marshal handles the escape of newlines / quotes / unicode so
|
||
// the agent's reply round-trips intact.
|
||
contentBytes, err := json.Marshal(map[string]string{"text": p.Text})
|
||
if err != nil {
|
||
return "", fmt.Errorf("lark http client: encode text content: %w", err)
|
||
}
|
||
path, body := outboundMessageRequest(p.ChatID, "text", string(contentBytes), p.ReplyTarget)
|
||
var resp struct {
|
||
Code int `json:"code"`
|
||
Msg string `json:"msg"`
|
||
Data struct {
|
||
MessageID string `json:"message_id"`
|
||
} `json:"data"`
|
||
}
|
||
if err := c.doJSON(ctx, c.resolveBaseURL(p.InstallationID), http.MethodPost, path, token, body, &resp); err != nil {
|
||
return "", fmt.Errorf("lark http client: send text message: %w", err)
|
||
}
|
||
if resp.Code != 0 || resp.Data.MessageID == "" {
|
||
if isTokenError(resp.Code) {
|
||
c.invalidateToken(p.InstallationID.AppID)
|
||
}
|
||
return "", &APIError{Op: "send text message", Code: resp.Code, Msg: resp.Msg}
|
||
}
|
||
return resp.Data.MessageID, nil
|
||
}
|
||
|
||
// SendMarkdownCard posts the agent's reply as an interactive card
|
||
// using Lark's schema-2.0 envelope with a single `tag: "markdown"`
|
||
// body element. Lark's client renders the markdown into formatted
|
||
// text (bold, italics, lists, links, fenced code blocks, tables, …)
|
||
// rather than showing raw markdown characters as it does for
|
||
// `msg_type=text`. We deliberately keep `SendTextMessage` as a
|
||
// separate path for plain-prose replies — a card around a one-line
|
||
// "Hello!" adds visual chrome that the user doesn't want; the
|
||
// routing decision (markdown vs text) lives at the Patcher layer.
|
||
//
|
||
// Why schema 2.0 rather than the legacy schema with a `div` +
|
||
// `lark_md` text element: the legacy `lark_md` tag's markdown
|
||
// dialect is much narrower — no fenced code blocks (syntax
|
||
// highlighting), no tables, no heading sizes. Schema-2.0's
|
||
// `markdown` tag is closer to GFM.
|
||
func (c *httpAPIClient) SendMarkdownCard(ctx context.Context, p SendMarkdownCardParams) (string, error) {
|
||
if p.ChatID == "" {
|
||
return "", errors.New("lark http client: missing chat_id")
|
||
}
|
||
if p.Markdown == "" {
|
||
return "", errors.New("lark http client: missing markdown body")
|
||
}
|
||
token, err := c.tenantAccessToken(ctx, p.InstallationID)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
card := map[string]any{
|
||
"schema": "2.0",
|
||
"body": map[string]any{
|
||
"elements": []any{
|
||
map[string]any{"tag": "markdown", "content": p.Markdown},
|
||
},
|
||
},
|
||
}
|
||
if p.Summary != "" {
|
||
card["config"] = map[string]any{
|
||
"summary": map[string]any{"content": p.Summary},
|
||
}
|
||
}
|
||
cardBytes, err := json.Marshal(card)
|
||
if err != nil {
|
||
return "", fmt.Errorf("lark http client: encode markdown card: %w", err)
|
||
}
|
||
path, body := outboundMessageRequest(p.ChatID, "interactive", string(cardBytes), p.ReplyTarget)
|
||
var resp struct {
|
||
Code int `json:"code"`
|
||
Msg string `json:"msg"`
|
||
Data struct {
|
||
MessageID string `json:"message_id"`
|
||
} `json:"data"`
|
||
}
|
||
if err := c.doJSON(ctx, c.resolveBaseURL(p.InstallationID), http.MethodPost, path, token, body, &resp); err != nil {
|
||
return "", fmt.Errorf("lark http client: send markdown card: %w", err)
|
||
}
|
||
if resp.Code != 0 || resp.Data.MessageID == "" {
|
||
if isTokenError(resp.Code) {
|
||
c.invalidateToken(p.InstallationID.AppID)
|
||
}
|
||
return "", &APIError{Op: "send markdown card", Code: resp.Code, Msg: resp.Msg}
|
||
}
|
||
return resp.Data.MessageID, nil
|
||
}
|
||
|
||
// PatchInteractiveCard updates an existing card's body. Lark's
|
||
// message-patch endpoint replaces the whole card payload; callers
|
||
// (i.e. the Patcher) render the full updated card each time.
|
||
func (c *httpAPIClient) PatchInteractiveCard(ctx context.Context, p PatchCardParams) error {
|
||
if p.LarkCardMessageID == "" {
|
||
return errors.New("lark http client: missing card message id")
|
||
}
|
||
if p.CardJSON == "" {
|
||
return errors.New("lark http client: missing card json")
|
||
}
|
||
token, err := c.tenantAccessToken(ctx, p.InstallationID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
body := map[string]string{"content": p.CardJSON}
|
||
var resp struct {
|
||
Code int `json:"code"`
|
||
Msg string `json:"msg"`
|
||
}
|
||
path := "/open-apis/im/v1/messages/" + url.PathEscape(p.LarkCardMessageID)
|
||
if err := c.doJSON(ctx, c.resolveBaseURL(p.InstallationID), http.MethodPatch, path, token, body, &resp); err != nil {
|
||
return fmt.Errorf("lark http client: patch interactive card: %w", err)
|
||
}
|
||
if resp.Code != 0 {
|
||
if isTokenError(resp.Code) {
|
||
c.invalidateToken(p.InstallationID.AppID)
|
||
}
|
||
return fmt.Errorf("lark http client: patch interactive card: code=%d msg=%q", resp.Code, resp.Msg)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// SendBindingPromptCard renders the member-binding card and posts it
|
||
// directly to the unbound user's open_id (not the chat). Keeping the
|
||
// card template inside this client — rather than the dispatcher —
|
||
// means the dispatcher never has to know about Lark's card schema.
|
||
func (c *httpAPIClient) SendBindingPromptCard(ctx context.Context, p BindingPromptParams) error {
|
||
if p.OpenID == "" {
|
||
return errors.New("lark http client: missing open_id")
|
||
}
|
||
if p.BindURL == "" {
|
||
return errors.New("lark http client: missing bind url")
|
||
}
|
||
cardJSON, err := bindingPromptTemplate(p.BindURL)
|
||
if err != nil {
|
||
return fmt.Errorf("lark http client: render binding prompt: %w", err)
|
||
}
|
||
token, err := c.tenantAccessToken(ctx, p.InstallationID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
q := url.Values{}
|
||
q.Set("receive_id_type", "open_id")
|
||
body := map[string]string{
|
||
"receive_id": string(p.OpenID),
|
||
"msg_type": "interactive",
|
||
"content": cardJSON,
|
||
}
|
||
var resp struct {
|
||
Code int `json:"code"`
|
||
Msg string `json:"msg"`
|
||
}
|
||
path := "/open-apis/im/v1/messages?" + q.Encode()
|
||
if err := c.doJSON(ctx, c.resolveBaseURL(p.InstallationID), http.MethodPost, path, token, body, &resp); err != nil {
|
||
return fmt.Errorf("lark http client: send binding prompt: %w", err)
|
||
}
|
||
if resp.Code != 0 {
|
||
if isTokenError(resp.Code) {
|
||
c.invalidateToken(p.InstallationID.AppID)
|
||
}
|
||
return fmt.Errorf("lark http client: send binding prompt: code=%d msg=%q", resp.Code, resp.Msg)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// GetBotInfo calls /open-apis/bot/v3/info to learn the Bot's
|
||
// per-installation `open_id` and then /open-apis/contact/v3/users/
|
||
// {open_id}?user_id_type=open_id to resolve its stable `union_id`.
|
||
// RegistrationService is the only caller — right after the device-
|
||
// flow registration returns fresh `client_id` / `client_secret`, the
|
||
// service mints a tenant_access_token with those creds and calls
|
||
// this method so the installation row can be frozen with both Bot
|
||
// identifiers in the same transaction as the installer-bind.
|
||
//
|
||
// Why two API calls instead of one: /bot/v3/info does not return
|
||
// union_id in the public schema. The WS inbound decoder needs
|
||
// union_id to disambiguate which bot was @-mentioned in a multi-bot
|
||
// group chat (the per-app open_id field on mentions is structurally
|
||
// inverse across WS perspectives — see MUL-2671 triage), so we
|
||
// invest one extra HTTP round-trip at install time to capture it
|
||
// and avoid running the wrong supervisor for every event going
|
||
// forward.
|
||
//
|
||
// A missing union_id (contact lookup denied by app scope, or Lark
|
||
// returns an empty field) is NOT a hard failure here — the
|
||
// installation is still usable for p2p chats and the decoder can
|
||
// fall back to the (broken) open_id match path until the operator
|
||
// fixes scopes. We log a warning so the gap is visible.
|
||
//
|
||
// Other fields the upstream APIs return (display name, avatar, IP
|
||
// whitelist) are deliberately dropped; downstream reads can fetch
|
||
// them on demand from the bot_open_id, and freezing them into our
|
||
// schema would create a drift surface every time the operator edits
|
||
// the Bot on Lark's side.
|
||
func (c *httpAPIClient) GetBotInfo(ctx context.Context, creds InstallationCredentials) (BotInfo, error) {
|
||
if creds.AppID == "" || creds.AppSecret == "" {
|
||
return BotInfo{}, errors.New("lark http client: missing app credentials for GetBotInfo")
|
||
}
|
||
token, err := c.tenantAccessToken(ctx, creds)
|
||
if err != nil {
|
||
return BotInfo{}, err
|
||
}
|
||
var botResp struct {
|
||
Code int `json:"code"`
|
||
Msg string `json:"msg"`
|
||
Bot struct {
|
||
OpenID string `json:"open_id"`
|
||
} `json:"bot"`
|
||
}
|
||
if err := c.doJSON(ctx, c.resolveBaseURL(creds), http.MethodGet, "/open-apis/bot/v3/info", token, nil, &botResp); err != nil {
|
||
return BotInfo{}, fmt.Errorf("lark http client: bot info: %w", err)
|
||
}
|
||
if botResp.Code != 0 {
|
||
if isTokenError(botResp.Code) {
|
||
c.invalidateToken(creds.AppID)
|
||
}
|
||
return BotInfo{}, fmt.Errorf("lark http client: bot info: code=%d msg=%q", botResp.Code, botResp.Msg)
|
||
}
|
||
if botResp.Bot.OpenID == "" {
|
||
return BotInfo{}, errors.New("lark http client: bot info: response missing open_id")
|
||
}
|
||
|
||
// Resolve union_id via the contact endpoint. Soft-fail: log and
|
||
// return the BotInfo with empty UnionID. Callers (Registration-
|
||
// Service.finishSuccess) accept the gap and persist what they
|
||
// have.
|
||
unionID, lookupErr := c.fetchBotUnionID(ctx, c.resolveBaseURL(creds), creds.AppID, token, botResp.Bot.OpenID)
|
||
if lookupErr != nil {
|
||
c.cfg.Logger.Warn("lark http client: bot union_id lookup failed; continuing without it",
|
||
"app_id", creds.AppID,
|
||
"bot_open_id", botResp.Bot.OpenID,
|
||
"err", lookupErr)
|
||
}
|
||
return BotInfo{OpenID: OpenID(botResp.Bot.OpenID), UnionID: unionID}, nil
|
||
}
|
||
|
||
// GetMessage retrieves a message by id via
|
||
// GET /open-apis/im/v1/messages/{message_id}. The endpoint always wraps
|
||
// the result in data.items[] — one element for a normal message, and a
|
||
// forward sentinel followed by the bundled child messages for a
|
||
// `merge_forward`. We pass user_id_type=open_id so sender.id and
|
||
// mentions[].id come back as open_ids, matching the identifiers the
|
||
// rest of the package keys on.
|
||
//
|
||
// body.content is forwarded verbatim (the raw, JSON-encoded, msg_type-
|
||
// specific string Lark double-encodes); the enricher's flattener owns
|
||
// interpreting it. A deleted / out-of-scope message surfaces as a Lark
|
||
// error code, which we turn into a normal Go error so the enricher can
|
||
// degrade to its "[unable to fetch]" placeholder without aborting the
|
||
// inbound pipeline.
|
||
func (c *httpAPIClient) GetMessage(ctx context.Context, creds InstallationCredentials, messageID string) ([]LarkMessage, error) {
|
||
if messageID == "" {
|
||
return nil, errors.New("lark http client: missing message_id")
|
||
}
|
||
token, err := c.tenantAccessToken(ctx, creds)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
q := url.Values{}
|
||
q.Set("user_id_type", "open_id")
|
||
path := "/open-apis/im/v1/messages/" + url.PathEscape(messageID) + "?" + q.Encode()
|
||
|
||
var resp struct {
|
||
Code int `json:"code"`
|
||
Msg string `json:"msg"`
|
||
Data struct {
|
||
Items []larkRESTMessageItem `json:"items"`
|
||
} `json:"data"`
|
||
}
|
||
if err := c.doJSON(ctx, c.resolveBaseURL(creds), http.MethodGet, path, token, nil, &resp); err != nil {
|
||
return nil, fmt.Errorf("lark http client: get message: %w", err)
|
||
}
|
||
if resp.Code != 0 {
|
||
if isTokenError(resp.Code) {
|
||
c.invalidateToken(creds.AppID)
|
||
}
|
||
return nil, fmt.Errorf("lark http client: get message: code=%d msg=%q", resp.Code, resp.Msg)
|
||
}
|
||
|
||
out := make([]LarkMessage, 0, len(resp.Data.Items))
|
||
for _, it := range resp.Data.Items {
|
||
out = append(out, it.normalize())
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// larkListMessagesMaxPageSize is Lark's hard cap on a single
|
||
// im/v1/messages page. We clamp to it so a caller asking for more
|
||
// silently gets the max rather than a 400 from Lark.
|
||
const larkListMessagesMaxPageSize = 50
|
||
|
||
// ListChatMessages retrieves a bounded, recent window of messages via
|
||
// GET /open-apis/im/v1/messages. Where GetMessage fetches a single message
|
||
// by id, this lists a conversation; it backs the enricher's group-context
|
||
// prefetch. The container is chat (container_id_type=chat) by default, or a
|
||
// single Lark topic (container_id_type=thread) when p.ThreadID is set — the
|
||
// thread container keeps a topic @-mention from seeing sibling topics that
|
||
// share the chat_id (#5835). We pass sort_type=ByCreateTimeDesc so the
|
||
// newest messages come first and a small page_size captures "the last N"
|
||
// without paginating, keeping the inbound ACK path's fan-out to a single
|
||
// round-trip. user_id_type=open_id matches the identifiers the rest of the
|
||
// package keys on; body.content is forwarded verbatim for the enricher's
|
||
// flattener to interpret.
|
||
func (c *httpAPIClient) ListChatMessages(ctx context.Context, creds InstallationCredentials, p ListMessagesParams) ([]LarkMessage, error) {
|
||
if p.ChatID == "" {
|
||
return nil, errors.New("lark http client: missing chat_id")
|
||
}
|
||
size := p.PageSize
|
||
if size <= 0 {
|
||
size = 1
|
||
} else if size > larkListMessagesMaxPageSize {
|
||
size = larkListMessagesMaxPageSize
|
||
}
|
||
token, err := c.tenantAccessToken(ctx, creds)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
q := url.Values{}
|
||
if p.ThreadID != "" {
|
||
// Topic-scoped window: only this 话题's messages, so a @-mention
|
||
// inside a topic never pulls sibling topics that share the chat_id
|
||
// (#5835). The thread container rejects end_time, so it is omitted
|
||
// here; the caller anchors the window to the trigger time
|
||
// client-side instead.
|
||
q.Set("container_id_type", "thread")
|
||
q.Set("container_id", p.ThreadID)
|
||
} else {
|
||
q.Set("container_id_type", "chat")
|
||
q.Set("container_id", string(p.ChatID))
|
||
if p.EndTime > 0 {
|
||
q.Set("end_time", strconv.FormatInt(p.EndTime, 10))
|
||
}
|
||
}
|
||
q.Set("sort_type", "ByCreateTimeDesc")
|
||
q.Set("page_size", strconv.Itoa(size))
|
||
q.Set("user_id_type", "open_id")
|
||
path := "/open-apis/im/v1/messages?" + q.Encode()
|
||
|
||
var resp struct {
|
||
Code int `json:"code"`
|
||
Msg string `json:"msg"`
|
||
Data struct {
|
||
Items []larkRESTMessageItem `json:"items"`
|
||
} `json:"data"`
|
||
}
|
||
if err := c.doJSON(ctx, c.resolveBaseURL(creds), http.MethodGet, path, token, nil, &resp); err != nil {
|
||
return nil, fmt.Errorf("lark http client: list chat messages: %w", err)
|
||
}
|
||
if resp.Code != 0 {
|
||
if isTokenError(resp.Code) {
|
||
c.invalidateToken(creds.AppID)
|
||
}
|
||
return nil, fmt.Errorf("lark http client: list chat messages: code=%d msg=%q", resp.Code, resp.Msg)
|
||
}
|
||
|
||
out := make([]LarkMessage, 0, len(resp.Data.Items))
|
||
for _, it := range resp.Data.Items {
|
||
out = append(out, it.normalize())
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// DownloadMessageResource obtains a binary message resource (image, video,
|
||
// file, audio) from Lark/Feishu. Business errors are still represented as
|
||
// JSON with a code/msg body on some failures, so JSON-looking responses are
|
||
// checked before being treated as resource bytes.
|
||
func (c *httpAPIClient) DownloadMessageResource(ctx context.Context, creds InstallationCredentials, p DownloadResourceParams) (DownloadedResource, error) {
|
||
stream, err := c.DownloadMessageResourceStream(ctx, creds, p)
|
||
if err != nil {
|
||
return DownloadedResource{}, err
|
||
}
|
||
defer stream.Body.Close()
|
||
rawBody, err := io.ReadAll(stream.Body)
|
||
if err != nil {
|
||
return DownloadedResource{}, fmt.Errorf("lark http client: download resource: read body: %w", err)
|
||
}
|
||
sizeBytes := stream.SizeBytes
|
||
if sizeBytes == 0 {
|
||
sizeBytes = int64(len(rawBody))
|
||
}
|
||
return DownloadedResource{
|
||
Data: rawBody,
|
||
ContentType: stream.ContentType,
|
||
Filename: stream.Filename,
|
||
SizeBytes: sizeBytes,
|
||
}, nil
|
||
}
|
||
|
||
func (c *httpAPIClient) DownloadMessageResourceStream(ctx context.Context, creds InstallationCredentials, p DownloadResourceParams) (DownloadedResourceStream, error) {
|
||
if p.MessageID == "" {
|
||
return DownloadedResourceStream{}, errors.New("lark http client: missing message_id")
|
||
}
|
||
if p.FileKey == "" {
|
||
return DownloadedResourceStream{}, errors.New("lark http client: missing file_key")
|
||
}
|
||
token, err := c.tenantAccessToken(ctx, creds)
|
||
if err != nil {
|
||
return DownloadedResourceStream{}, err
|
||
}
|
||
q := url.Values{}
|
||
if p.Type != "" {
|
||
q.Set("type", p.Type)
|
||
}
|
||
path := "/open-apis/im/v1/messages/" + url.PathEscape(p.MessageID) + "/resources/" + url.PathEscape(p.FileKey)
|
||
if encoded := q.Encode(); encoded != "" {
|
||
path += "?" + encoded
|
||
}
|
||
|
||
downloadCtx := ctx
|
||
var cancel context.CancelFunc
|
||
if c.cfg.ResourceDownloadTimeout > 0 {
|
||
downloadCtx, cancel = context.WithTimeout(ctx, c.cfg.ResourceDownloadTimeout)
|
||
}
|
||
req, err := http.NewRequestWithContext(downloadCtx, http.MethodGet, c.resolveBaseURL(creds)+path, nil)
|
||
if err != nil {
|
||
if cancel != nil {
|
||
cancel()
|
||
}
|
||
return DownloadedResourceStream{}, fmt.Errorf("lark http client: download resource: new request: %w", err)
|
||
}
|
||
req.Header.Set("Authorization", "Bearer "+token)
|
||
resp, err := c.cfg.ResourceHTTPClient.Do(req)
|
||
if err != nil {
|
||
if cancel != nil {
|
||
cancel()
|
||
}
|
||
return DownloadedResourceStream{}, fmt.Errorf("lark http client: download resource: http do: %w", err)
|
||
}
|
||
closeWithCancel := func() {
|
||
resp.Body.Close()
|
||
if cancel != nil {
|
||
cancel()
|
||
}
|
||
}
|
||
if resp.ContentLength > maxMessageResourceBytes {
|
||
closeWithCancel()
|
||
return DownloadedResourceStream{}, fmt.Errorf("lark http client: download resource: resource exceeds %d bytes", maxMessageResourceBytes)
|
||
}
|
||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||
rawBody, readErr := readMessageResourceErrorBody(resp.Body)
|
||
closeWithCancel()
|
||
if readErr != nil {
|
||
return DownloadedResourceStream{}, readErr
|
||
}
|
||
return DownloadedResourceStream{}, fmt.Errorf("lark http client: download resource: http %d: %s", resp.StatusCode, truncate(string(rawBody), 512))
|
||
}
|
||
contentType := resp.Header.Get("Content-Type")
|
||
if strings.Contains(strings.ToLower(contentType), "json") {
|
||
rawBody, readErr := readMessageResourceErrorBody(resp.Body)
|
||
closeWithCancel()
|
||
if readErr != nil {
|
||
return DownloadedResourceStream{}, readErr
|
||
}
|
||
var apiResp struct {
|
||
Code int `json:"code"`
|
||
Msg string `json:"msg"`
|
||
}
|
||
if err := json.Unmarshal(rawBody, &apiResp); err == nil && apiResp.Code != 0 {
|
||
if isTokenError(apiResp.Code) {
|
||
c.invalidateToken(creds.AppID)
|
||
}
|
||
return DownloadedResourceStream{}, &APIError{Op: "download resource", Code: apiResp.Code, Msg: apiResp.Msg}
|
||
}
|
||
return DownloadedResourceStream{
|
||
Body: cancelOnClose(io.NopCloser(bytes.NewReader(rawBody)), cancel),
|
||
ContentType: contentType,
|
||
Filename: filenameFromContentDisposition(resp.Header.Get("Content-Disposition")),
|
||
SizeBytes: int64(len(rawBody)),
|
||
}, nil
|
||
}
|
||
if contentType == "" {
|
||
contentType = "application/octet-stream"
|
||
}
|
||
sizeBytes := resp.ContentLength
|
||
if sizeBytes < 0 {
|
||
sizeBytes = 0
|
||
}
|
||
return DownloadedResourceStream{
|
||
Body: cancelOnClose(&maxBytesReadCloser{r: resp.Body, remaining: maxMessageResourceBytes}, cancel),
|
||
ContentType: contentType,
|
||
Filename: filenameFromContentDisposition(resp.Header.Get("Content-Disposition")),
|
||
SizeBytes: sizeBytes,
|
||
}, nil
|
||
}
|
||
|
||
func readMessageResourceErrorBody(body io.Reader) ([]byte, error) {
|
||
rawBody, err := io.ReadAll(io.LimitReader(body, maxMessageResourceBytes+1))
|
||
if err != nil {
|
||
return nil, fmt.Errorf("lark http client: download resource: read body: %w", err)
|
||
}
|
||
if len(rawBody) > maxMessageResourceBytes {
|
||
return nil, fmt.Errorf("lark http client: download resource: resource exceeds %d bytes", maxMessageResourceBytes)
|
||
}
|
||
return rawBody, nil
|
||
}
|
||
|
||
type maxBytesReadCloser struct {
|
||
r io.ReadCloser
|
||
remaining int64
|
||
}
|
||
|
||
func (r *maxBytesReadCloser) Read(p []byte) (int, error) {
|
||
if r.remaining > 0 {
|
||
if int64(len(p)) > r.remaining {
|
||
p = p[:r.remaining]
|
||
}
|
||
n, err := r.r.Read(p)
|
||
r.remaining -= int64(n)
|
||
return n, err
|
||
}
|
||
var one [1]byte
|
||
n, err := r.r.Read(one[:])
|
||
if n > 0 {
|
||
return 0, fmt.Errorf("lark http client: download resource: resource exceeds %d bytes", maxMessageResourceBytes)
|
||
}
|
||
return 0, err
|
||
}
|
||
|
||
func (r *maxBytesReadCloser) Close() error {
|
||
return r.r.Close()
|
||
}
|
||
|
||
type cancelReadCloser struct {
|
||
io.ReadCloser
|
||
cancel context.CancelFunc
|
||
}
|
||
|
||
func cancelOnClose(body io.ReadCloser, cancel context.CancelFunc) io.ReadCloser {
|
||
if cancel == nil {
|
||
return body
|
||
}
|
||
return &cancelReadCloser{ReadCloser: body, cancel: cancel}
|
||
}
|
||
|
||
func (r *cancelReadCloser) Close() error {
|
||
err := r.ReadCloser.Close()
|
||
r.cancel()
|
||
return err
|
||
}
|
||
|
||
func filenameFromContentDisposition(raw string) string {
|
||
if raw == "" {
|
||
return ""
|
||
}
|
||
_, params, err := mime.ParseMediaType(raw)
|
||
if err != nil {
|
||
return ""
|
||
}
|
||
return params["filename"]
|
||
}
|
||
|
||
// larkBatchGetUsersMaxIDs is Lark's hard cap on user_ids per
|
||
// contact/v3/users/batch call. We drop the overflow rather than error so
|
||
// a caller asking for more still gets the first 50 resolved.
|
||
const larkBatchGetUsersMaxIDs = 50
|
||
|
||
// AddMessageReaction adds an emoji reaction to a message via
|
||
// POST /open-apis/im/v1/messages/{message_id}/reactions.
|
||
// Returns the reaction_id so it can be deleted later.
|
||
func (c *httpAPIClient) AddMessageReaction(ctx context.Context, p AddReactionParams) (string, error) {
|
||
if p.MessageID == "" {
|
||
return "", errors.New("lark http client: missing message_id")
|
||
}
|
||
if p.EmojiType == "" {
|
||
return "", errors.New("lark http client: missing emoji_type")
|
||
}
|
||
token, err := c.tenantAccessToken(ctx, p.InstallationID)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
body := map[string]any{
|
||
"reaction_type": map[string]string{"emoji_type": p.EmojiType},
|
||
}
|
||
path := "/open-apis/im/v1/messages/" + url.PathEscape(p.MessageID) + "/reactions"
|
||
var resp struct {
|
||
Code int `json:"code"`
|
||
Msg string `json:"msg"`
|
||
Data struct {
|
||
ReactionID string `json:"reaction_id"`
|
||
} `json:"data"`
|
||
}
|
||
if err := c.doJSON(ctx, c.resolveBaseURL(p.InstallationID), http.MethodPost, path, token, body, &resp); err != nil {
|
||
return "", fmt.Errorf("lark http client: add message reaction: %w", err)
|
||
}
|
||
if resp.Code != 0 || resp.Data.ReactionID == "" {
|
||
if isTokenError(resp.Code) {
|
||
c.invalidateToken(p.InstallationID.AppID)
|
||
}
|
||
return "", fmt.Errorf("lark http client: add message reaction: code=%d msg=%q", resp.Code, resp.Msg)
|
||
}
|
||
return resp.Data.ReactionID, nil
|
||
}
|
||
|
||
// DeleteMessageReaction removes a reaction from a message via
|
||
// DELETE /open-apis/im/v1/messages/{message_id}/reactions/{reaction_id}.
|
||
func (c *httpAPIClient) DeleteMessageReaction(ctx context.Context, p DeleteReactionParams) error {
|
||
if p.MessageID == "" {
|
||
return errors.New("lark http client: missing message_id")
|
||
}
|
||
if p.ReactionID == "" {
|
||
return errors.New("lark http client: missing reaction_id")
|
||
}
|
||
token, err := c.tenantAccessToken(ctx, p.InstallationID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
path := "/open-apis/im/v1/messages/" + url.PathEscape(p.MessageID) + "/reactions/" + url.PathEscape(p.ReactionID)
|
||
var resp struct {
|
||
Code int `json:"code"`
|
||
Msg string `json:"msg"`
|
||
}
|
||
if err := c.doJSON(ctx, c.resolveBaseURL(p.InstallationID), http.MethodDelete, path, token, nil, &resp); err != nil {
|
||
return fmt.Errorf("lark http client: delete message reaction: %w", err)
|
||
}
|
||
if resp.Code != 0 {
|
||
if isTokenError(resp.Code) {
|
||
c.invalidateToken(p.InstallationID.AppID)
|
||
}
|
||
return fmt.Errorf("lark http client: delete message reaction: code=%d msg=%q", resp.Code, resp.Msg)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// BatchGetUsers resolves user open_ids to display names via
|
||
// GET /open-apis/contact/v3/users/batch?user_ids=…&user_id_type=open_id.
|
||
// It mirrors fetchBotUnionID's single-user contact lookup, batched. Only
|
||
// id->name pairs the API actually returns are included; a restricted
|
||
// contact scope or an unknown id simply yields a smaller map (code==0
|
||
// with fewer items), never an error, so the enricher degrades to
|
||
// positional speaker labels. Ids past Lark's 50-per-call cap are dropped.
|
||
func (c *httpAPIClient) BatchGetUsers(ctx context.Context, creds InstallationCredentials, openIDs []string) (map[string]string, error) {
|
||
if len(openIDs) == 0 {
|
||
return map[string]string{}, nil
|
||
}
|
||
if len(openIDs) > larkBatchGetUsersMaxIDs {
|
||
openIDs = openIDs[:larkBatchGetUsersMaxIDs]
|
||
}
|
||
token, err := c.tenantAccessToken(ctx, creds)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
q := url.Values{}
|
||
q.Set("user_id_type", "open_id")
|
||
for _, id := range openIDs {
|
||
if id != "" {
|
||
q.Add("user_ids", id)
|
||
}
|
||
}
|
||
path := "/open-apis/contact/v3/users/batch?" + q.Encode()
|
||
|
||
var resp struct {
|
||
Code int `json:"code"`
|
||
Msg string `json:"msg"`
|
||
Data struct {
|
||
Items []struct {
|
||
OpenID string `json:"open_id"`
|
||
Name string `json:"name"`
|
||
} `json:"items"`
|
||
} `json:"data"`
|
||
}
|
||
if err := c.doJSON(ctx, c.resolveBaseURL(creds), http.MethodGet, path, token, nil, &resp); err != nil {
|
||
return nil, fmt.Errorf("lark http client: batch get users: %w", err)
|
||
}
|
||
if resp.Code != 0 {
|
||
if isTokenError(resp.Code) {
|
||
c.invalidateToken(creds.AppID)
|
||
}
|
||
return nil, fmt.Errorf("lark http client: batch get users: code=%d msg=%q", resp.Code, resp.Msg)
|
||
}
|
||
|
||
out := make(map[string]string, len(resp.Data.Items))
|
||
for _, it := range resp.Data.Items {
|
||
if it.OpenID != "" && it.Name != "" {
|
||
out[it.OpenID] = it.Name
|
||
}
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// larkRESTMessageItem is the IM v1 message item shape returned by the
|
||
// get / list endpoints. It differs from the WS receive event in two
|
||
// ways the enricher cares about: msg_type (not message_type), and a
|
||
// flat `sender.id` / `mentions[].id` string (not a nested id object).
|
||
type larkRESTMessageItem struct {
|
||
MessageID string `json:"message_id"`
|
||
RootID string `json:"root_id"`
|
||
ParentID string `json:"parent_id"`
|
||
ThreadID string `json:"thread_id"`
|
||
UpperMessageID string `json:"upper_message_id"`
|
||
MsgType string `json:"msg_type"`
|
||
CreateTime string `json:"create_time"`
|
||
Deleted bool `json:"deleted"`
|
||
Sender struct {
|
||
ID string `json:"id"`
|
||
IDType string `json:"id_type"`
|
||
SenderType string `json:"sender_type"`
|
||
} `json:"sender"`
|
||
Body struct {
|
||
Content string `json:"content"`
|
||
} `json:"body"`
|
||
Mentions []struct {
|
||
Key string `json:"key"`
|
||
ID string `json:"id"`
|
||
Name string `json:"name"`
|
||
} `json:"mentions"`
|
||
}
|
||
|
||
func (it larkRESTMessageItem) normalize() LarkMessage {
|
||
m := LarkMessage{
|
||
MessageID: it.MessageID,
|
||
MessageType: it.MsgType,
|
||
Content: it.Body.Content,
|
||
SenderID: it.Sender.ID,
|
||
SenderType: it.Sender.SenderType,
|
||
CreateTime: it.CreateTime,
|
||
ParentID: it.ParentID,
|
||
RootID: it.RootID,
|
||
ThreadID: it.ThreadID,
|
||
UpperMessageID: it.UpperMessageID,
|
||
Deleted: it.Deleted,
|
||
}
|
||
for _, mn := range it.Mentions {
|
||
m.Mentions = append(m.Mentions, LarkMessageMention{Key: mn.Key, ID: mn.ID, Name: mn.Name})
|
||
}
|
||
return m
|
||
}
|
||
|
||
// fetchBotUnionID resolves a Bot's `union_id` from its `open_id` via
|
||
// /open-apis/contact/v3/users/{open_id}?user_id_type=open_id. Split
|
||
// out from GetBotInfo so the failure mode is explicit and the call
|
||
// sites that only need open_id don't pay for the second round-trip.
|
||
//
|
||
// Empty string + nil error is a valid outcome: Lark's user endpoint
|
||
// can return code=0 with no union_id field when the app's contact
|
||
// scope is restricted. Caller logs and continues; the decoder still
|
||
// works in single-bot deployments where open_id-based matching is
|
||
// unambiguous.
|
||
func (c *httpAPIClient) fetchBotUnionID(ctx context.Context, baseURL, appID, token, openID string) (string, error) {
|
||
if openID == "" {
|
||
return "", errors.New("empty open_id")
|
||
}
|
||
q := url.Values{}
|
||
q.Set("user_id_type", "open_id")
|
||
path := "/open-apis/contact/v3/users/" + url.PathEscape(openID) + "?" + q.Encode()
|
||
var resp struct {
|
||
Code int `json:"code"`
|
||
Msg string `json:"msg"`
|
||
Data struct {
|
||
User struct {
|
||
UnionID string `json:"union_id"`
|
||
} `json:"user"`
|
||
} `json:"data"`
|
||
}
|
||
if err := c.doJSON(ctx, baseURL, http.MethodGet, path, token, nil, &resp); err != nil {
|
||
return "", fmt.Errorf("contact users: %w", err)
|
||
}
|
||
if resp.Code != 0 {
|
||
// invalidateToken is keyed by app_id (the cache key on
|
||
// httpAPIClient.tokens), NOT by the bearer string. Passing
|
||
// the bearer would do nothing and a stale token would keep
|
||
// being reused on every retry until natural TTL expiry.
|
||
if isTokenError(resp.Code) {
|
||
c.invalidateToken(appID)
|
||
}
|
||
return "", fmt.Errorf("contact users: code=%d msg=%q", resp.Code, resp.Msg)
|
||
}
|
||
return resp.Data.User.UnionID, nil
|
||
}
|
||
|
||
// doJSON encapsulates the verb + URL + auth-header + JSON
|
||
// encode/decode dance so each public method stays a thin shape-only
|
||
// adapter. baseURL is the per-call open-platform host the caller
|
||
// resolved via resolveBaseURL (region-aware). token == "" skips the
|
||
// Authorization header (only the tenant_access_token endpoint takes
|
||
// that path).
|
||
func (c *httpAPIClient) doJSON(ctx context.Context, baseURL, method, path, token string, body, out any) error {
|
||
var rdr io.Reader
|
||
if body != nil {
|
||
buf, err := json.Marshal(body)
|
||
if err != nil {
|
||
return fmt.Errorf("marshal body: %w", err)
|
||
}
|
||
rdr = bytes.NewReader(buf)
|
||
}
|
||
req, err := http.NewRequestWithContext(ctx, method, baseURL+path, rdr)
|
||
if err != nil {
|
||
return fmt.Errorf("new request: %w", err)
|
||
}
|
||
if body != nil {
|
||
req.Header.Set("Content-Type", "application/json; charset=utf-8")
|
||
}
|
||
if token != "" {
|
||
req.Header.Set("Authorization", "Bearer "+token)
|
||
}
|
||
resp, err := c.cfg.HTTPClient.Do(req)
|
||
if err != nil {
|
||
return fmt.Errorf("http do: %w", err)
|
||
}
|
||
defer resp.Body.Close()
|
||
rawBody, err := io.ReadAll(resp.Body)
|
||
if err != nil {
|
||
return fmt.Errorf("read body: %w", err)
|
||
}
|
||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||
return fmt.Errorf("http %d: %s", resp.StatusCode, truncate(string(rawBody), 512))
|
||
}
|
||
if out != nil && len(rawBody) > 0 {
|
||
if err := json.Unmarshal(rawBody, out); err != nil {
|
||
return fmt.Errorf("decode body: %w (raw=%s)", err, truncate(string(rawBody), 256))
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func isTokenError(code int) bool {
|
||
return code == codeTokenExpired || code == codeTokenInvalid
|
||
}
|
||
|
||
// APIError is a structured Lark business error: the request reached
|
||
// Lark, returned HTTP 200, but Lark rejected it with a non-zero
|
||
// `code`. This is distinct from the transport-level errors doJSON
|
||
// surfaces (network failure, 5xx, timeout), which are returned as
|
||
// plain wrapped errors. The distinction matters for the threaded-reply
|
||
// fallback: a business code is definitive ("nothing was sent, and here
|
||
// is exactly why"), whereas a transport error is ambiguous ("the
|
||
// message may or may not have been delivered") and must NOT trigger a
|
||
// chat-level retry that could duplicate or leak the reply.
|
||
type APIError struct {
|
||
Op string
|
||
Code int
|
||
Msg string
|
||
}
|
||
|
||
func (e *APIError) Error() string {
|
||
return fmt.Sprintf("lark http client: %s: code=%d msg=%q", e.Op, e.Code, e.Msg)
|
||
}
|
||
|
||
// threadReplyUnsupportedCodes are the reply-endpoint business codes
|
||
// that definitively mean "this specific trigger message / topic cannot
|
||
// receive a threaded reply" AND nothing was sent, while a plain
|
||
// chat-level send to the same chat is unaffected. Only these justify
|
||
// the chat-level fallback. Rate limits (230020), "message is being
|
||
// sent" (230049, ambiguous), permission/content errors (which would
|
||
// also fail at chat level), and all transport/5xx/timeout failures are
|
||
// deliberately excluded: those stay failures so we never duplicate a
|
||
// reply or leak a thread-only reply into the main group chat.
|
||
// Codes are from the IM reply-message endpoint error table.
|
||
var threadReplyUnsupportedCodes = map[int]struct{}{
|
||
230011: {}, // the trigger message has been recalled
|
||
230019: {}, // the topic does not exist
|
||
230050: {}, // the trigger message is invisible to the operator
|
||
230071: {}, // the group does not support reply in thread
|
||
230072: {}, // aggregated messages do not support reply in thread
|
||
230111: {}, // cannot reply to a self-destructing message
|
||
}
|
||
|
||
// isThreadReplyUnsupported reports whether err is a Lark APIError whose
|
||
// code means the threaded reply cannot land on this target. Only such
|
||
// errors are safe to retry at the chat level. Transport errors and
|
||
// other business codes return false.
|
||
func isThreadReplyUnsupported(err error) bool {
|
||
var apiErr *APIError
|
||
if errors.As(err, &apiErr) {
|
||
_, ok := threadReplyUnsupportedCodes[apiErr.Code]
|
||
return ok
|
||
}
|
||
return false
|
||
}
|
||
|
||
func truncate(s string, n int) string {
|
||
if len(s) <= n {
|
||
return s
|
||
}
|
||
return s[:n] + "…"
|
||
}
|
||
|
||
// bindingPromptTemplate renders the "you need to bind" interactive
|
||
// card. Single primary CTA pointing at the redemption URL; the rest
|
||
// of the body is plain-text Chinese copy matching the in-app voice.
|
||
//
|
||
// Kept here (not in defaultRenderer) so the binding card template can
|
||
// evolve independently of the streaming-status cards the Patcher
|
||
// renders — they have different lifecycles (binding card is one-shot,
|
||
// status cards are patched in place).
|
||
func bindingPromptTemplate(bindURL string) (string, error) {
|
||
doc := map[string]any{
|
||
"config": map[string]any{"wide_screen_mode": true},
|
||
"header": map[string]any{
|
||
"template": "blue",
|
||
"title": map[string]any{"tag": "plain_text", "content": "Multica"},
|
||
},
|
||
"elements": []any{
|
||
map[string]any{
|
||
"tag": "div",
|
||
"text": map[string]any{
|
||
"tag": "lark_md",
|
||
"content": "你还没有绑定 Multica 账户。点击下方按钮完成绑定后即可使用此 Agent。",
|
||
},
|
||
},
|
||
map[string]any{
|
||
"tag": "action",
|
||
"actions": []any{
|
||
map[string]any{
|
||
"tag": "button",
|
||
"text": map[string]any{"tag": "plain_text", "content": "去绑定"},
|
||
"type": "primary",
|
||
"url": bindURL,
|
||
},
|
||
},
|
||
},
|
||
},
|
||
}
|
||
raw, err := json.Marshal(doc)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
return string(raw), nil
|
||
}
|