package adapter import ( "context" "encoding/json" "errors" "fmt" "net/http" "time" ) // Telegram's own limits, per docs/adapter-contract.md. const ( textOnlyLimit = 4096 captionLimit = 1024 ) // recognisedConsentBases mirrors contracts/r1.openapi.yaml's subject_consent // enum. A post with none of these is refused: absence of objection is not // consent (FT-WP-0001 T07). var recognisedConsentBases = map[string]bool{ "explicit": true, "standing": true, "not-required": true, } // Telemetry is where the adapter emits what only it can see, per // docs/adapter-contract.md's "Telemetry the adapter should emit". It must // never carry subscriber identity or post body text. type Telemetry interface { Emit(ctx context.Context, event string, fields map[string]any) } // NopTelemetry discards events. Used when FLUID_CONTROL_URL is unset, e.g. in // tests and local runs. type NopTelemetry struct{} func (NopTelemetry) Emit(context.Context, string, map[string]any) {} // Config is what one adapter instance needs to publish to one channel. It is // derived, never hand-written: `provision adapter-config` produces it (see // internal/adaptercfg). One adapter instance serves one channel. type Config struct { ChatID int64 AttachVisual bool // R-2 (true) attaches a portrait; R-3 (false) is text-only HallBaseURL string } // Adapter implements the three R-1 operations. It never writes to Telegram // beyond sending to Config.ChatID, and it holds no MTProto session: see // docs/adapter-contract.md "What the adapter must not do". type Adapter struct { cfg Config sndr Sender store Store tel Telemetry now func() time.Time } // New builds an adapter for one channel. func New(cfg Config, sndr Sender, store Store, tel Telemetry) *Adapter { if tel == nil { tel = NopTelemetry{} } return &Adapter{cfg: cfg, sndr: sndr, store: store, tel: tel, now: time.Now} } // Handler wires the three endpoints exactly as contracts/r1.openapi.yaml // names them. The gateway validates requests against that contract before the // adapter sees them; this handler still assumes hostile content, per // docs/adapter-contract.md. func (a *Adapter) Handler() http.Handler { mux := http.NewServeMux() mux.HandleFunc("/v1/channel-posts", func(w http.ResponseWriter, r *http.Request) { switch r.Method { case http.MethodPost: a.handlePublish(w, r) case http.MethodGet: a.handleList(w, r) default: writeError(w, http.StatusMethodNotAllowed, "method not allowed") } }) mux.HandleFunc("/v1/channel-posts/", func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { writeError(w, http.StatusMethodNotAllowed, "method not allowed") return } postID := r.URL.Path[len("/v1/channel-posts/"):] if postID == "" { writeError(w, http.StatusNotFound, "post_id is required") return } a.handleGet(w, postID) }) return mux } // postRequest mirrors the requestBody schema in contracts/r1.openapi.yaml. type postRequest struct { PostID string `json:"post_id"` Body string `json:"body"` SourceEntryID string `json:"source_entry_id"` SourceEntryURL string `json:"source_entry_url"` ReviewedBy string `json:"reviewed_by"` SubjectDisplayName string `json:"subject_display_name,omitempty"` SubjectConsent string `json:"subject_consent,omitempty"` Variant string `json:"variant,omitempty"` VisualURL string `json:"visual_url,omitempty"` Tags []string `json:"tags,omitempty"` } func (a *Adapter) handlePublish(w http.ResponseWriter, r *http.Request) { var req postRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { writeError(w, http.StatusBadRequest, "malformed request body") return } if err := validateRequired(req); err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } // Refuse a post with no review or no recognised consent basis. Emitted as // guardrail telemetry so the series exists even though it should always // read false in practice, per docs/adapter-contract.md. if req.ReviewedBy == "" { a.tel.Emit(r.Context(), "unreviewed_publication", map[string]any{"post_id": req.PostID}) writeError(w, http.StatusForbidden, "reviewed_by is required") return } if req.SubjectConsent != "" && !recognisedConsentBases[req.SubjectConsent] { writeError(w, http.StatusBadRequest, fmt.Sprintf("subject_consent %q is not recognised", req.SubjectConsent)) return } if req.SubjectConsent == "" { a.tel.Emit(r.Context(), "missing_subject_consent", map[string]any{"post_id": req.PostID}) writeError(w, http.StatusForbidden, "no recognised consent basis for this subject's work") return } if req.SourceEntryID == "" || req.SourceEntryURL == "" { a.tel.Emit(r.Context(), "missing_source_attribution", map[string]any{"post_id": req.PostID}) writeError(w, http.StatusBadRequest, "source_entry_id and source_entry_url are required") return } limit := textOnlyLimit if a.cfg.AttachVisual { limit = captionLimit } if len(req.Body) > limit { // A composition that does not fit was not condensed; the adapter must // refuse rather than truncate (docs/adapter-contract.md). writeError(w, http.StatusBadRequest, fmt.Sprintf("body is %d characters, over the %d limit for this delivery form", len(req.Body), limit)) return } rendered, err := RenderHTML(req.Body) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } rendered += sourceLink(req.SourceEntryURL) if len(rendered) > limit { writeError(w, http.StatusBadRequest, "rendering plus source attribution exceeds the limit for this delivery form") return } existing, found := a.store.Get(req.PostID) if found { if existing.Body == rendered { writeJSON(w, http.StatusOK, publicationView(existing)) return } // Same post_id, changed content: edit rather than duplicate. A // recomposition of the same source entry is a NEW post_id, not this // path (docs/adapter-contract.md). if err := a.editExisting(r.Context(), existing, rendered, req); err != nil { a.storeFailure(req.PostID, existing, err) writeTelegramErr(w, err) return } existing.Body = rendered existing.Variant = req.Variant existing.SourceEntryID = req.SourceEntryID existing.PublishedAt = a.now() existing.State = StatePublished existing.LastAttemptErr = "" a.store.Put(existing) writeJSON(w, http.StatusOK, publicationView(existing)) return } msgID, err := a.sendNew(r.Context(), rendered, req) pub := Publication{ PostID: req.PostID, Body: rendered, SourceEntryID: req.SourceEntryID, Variant: req.Variant, PublishedAt: a.now(), } if err != nil { pub.State = StateFailed pub.LastAttemptErr = err.Error() a.store.Put(pub) writeTelegramErr(w, err) return } pub.State = StatePublished pub.TelegramMsgID = msgID a.store.Put(pub) a.tel.Emit(r.Context(), "published", map[string]any{"post_id": req.PostID, "variant": req.Variant}) writeJSON(w, http.StatusCreated, publicationView(pub)) } func (a *Adapter) sendNew(ctx context.Context, rendered string, req postRequest) (int, error) { if a.cfg.AttachVisual && req.VisualURL != "" { return a.sndr.SendPhoto(ctx, a.cfg.ChatID, req.VisualURL, rendered) } return a.sndr.SendMessage(ctx, a.cfg.ChatID, rendered) } func (a *Adapter) editExisting(ctx context.Context, existing Publication, rendered string, req postRequest) error { if a.cfg.AttachVisual && req.VisualURL != "" { return a.sndr.EditMessageCaption(ctx, a.cfg.ChatID, existing.TelegramMsgID, rendered) } return a.sndr.EditMessageText(ctx, a.cfg.ChatID, existing.TelegramMsgID, rendered) } func (a *Adapter) storeFailure(postID string, existing Publication, err error) { existing.LastAttemptErr = err.Error() a.store.Put(existing) _ = postID } func (a *Adapter) handleGet(w http.ResponseWriter, postID string) { pub, ok := a.store.Get(postID) if !ok { writeError(w, http.StatusNotFound, "this post has not been published") return } writeJSON(w, http.StatusOK, publicationView(pub)) } func (a *Adapter) handleList(w http.ResponseWriter, r *http.Request) { q := r.URL.Query() filter := ListFilter{ State: State(q.Get("state")), Variant: q.Get("variant"), SourceEntryID: q.Get("source_entry_id"), } if limit := q.Get("limit"); limit != "" { var n int if _, err := fmt.Sscanf(limit, "%d", &n); err == nil { filter.Limit = n } } pubs := a.store.List(filter) views := make([]map[string]any, 0, len(pubs)) for _, p := range pubs { views = append(views, publicationView(p)) } writeJSON(w, http.StatusOK, map[string]any{"posts": views}) } func validateRequired(req postRequest) error { if req.PostID == "" { return errors.New("post_id is required") } if len(req.PostID) > 128 { return errors.New("post_id exceeds 128 characters") } if req.Body == "" { return errors.New("body is required") } return nil } func sourceLink(url string) string { return fmt.Sprintf(` Source`, escapeAttr(url)) } func publicationView(p Publication) map[string]any { return map[string]any{ "post_id": p.PostID, "state": p.State, "source_entry_id": p.SourceEntryID, "variant": p.Variant, "published_at": p.PublishedAt, } } func writeJSON(w http.ResponseWriter, status int, v any) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) _ = json.NewEncoder(w).Encode(v) } func writeError(w http.ResponseWriter, status int, msg string) { writeJSON(w, status, map[string]string{"error": msg}) } // writeTelegramErr maps a Sender failure per docs/adapter-contract.md: a rate // limit or any other backend failure becomes 502, never retried automatically // for a call that may have already succeeded. func writeTelegramErr(w http.ResponseWriter, err error) { var rle *RateLimitError if errors.As(err, &rle) { w.Header().Set("Retry-After", fmt.Sprintf("%d", rle.RetryAfterSeconds)) } writeError(w, http.StatusBadGateway, "telegram was unreachable or refused the message") }