fluid-telegram/internal/adapter/adapter.go

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(`

<a href="%s">Source</a>`, 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")
}