vojo/apps/ai-bot/appservice.go

package main

import (
	"context"
	"crypto/subtle"
	"encoding/json"
	"log/slog"
	"net/http"
	"strings"
	"time"
)

// AppService is the homeserver-facing half of the bot: an HTTP server Synapse
// pushes transactions to (Application Service API). It authenticates every push
// with the hs_token, dedups by transaction id (idempotency, per spec), and hands
// the events to the bot's processing callback.
type AppService struct {
	cfg     *Config
	log     *slog.Logger
	store   *Store
	handler func(ctx context.Context, events []Event)
	baseCtx context.Context
}

func NewAppService(cfg *Config, logger *slog.Logger, store *Store, handler func(context.Context, []Event)) *AppService {
	return &AppService{cfg: cfg, log: logger, store: store, handler: handler}
}

// Serve starts the transaction server and blocks until ctx is cancelled.
func (a *AppService) Serve(ctx context.Context) error {
	a.baseCtx = ctx

	mux := http.NewServeMux()
	// Modern (/_matrix/app/v1) + legacy (unprefixed) paths — Synapse versions
	// differ on which they call.
	mux.HandleFunc("PUT /_matrix/app/v1/transactions/{txnId}", a.handleTransaction)
	mux.HandleFunc("PUT /transactions/{txnId}", a.handleTransaction)
	mux.HandleFunc("GET /_matrix/app/v1/users/{userId}", a.handleUserQuery)
	mux.HandleFunc("GET /users/{userId}", a.handleUserQuery)
	mux.HandleFunc("GET /_matrix/app/v1/rooms/{roomAlias}", a.handleRoomQuery)
	mux.HandleFunc("GET /rooms/{roomAlias}", a.handleRoomQuery)
	mux.HandleFunc("GET /", func(w http.ResponseWriter, _ *http.Request) { writeJSON(w, http.StatusOK, struct{}{}) })

	srv := &http.Server{
		Addr:              a.cfg.ASAddr,
		Handler:           mux,
		ReadHeaderTimeout: 10 * time.Second,
	}

	errCh := make(chan error, 1)
	go func() { errCh <- srv.ListenAndServe() }()
	a.log.Info("appservice listening", "addr", a.cfg.ASAddr)

	select {
	case <-ctx.Done():
		shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = srv.Shutdown(shutCtx)
		return nil
	case err := <-errCh:
		if err == http.ErrServerClosed {
			return nil
		}
		return err
	}
}

// authOK verifies the homeserver's hs_token (modern: Authorization: Bearer;
// legacy: ?access_token=). Constant-time compare to avoid token-timing leaks.
func (a *AppService) authOK(r *http.Request) bool {
	tok := ""
	if h := r.Header.Get("Authorization"); strings.HasPrefix(h, "Bearer ") {
		tok = strings.TrimPrefix(h, "Bearer ")
	} else {
		tok = r.URL.Query().Get("access_token")
	}
	return subtle.ConstantTimeCompare([]byte(tok), []byte(a.cfg.HSToken)) == 1
}

func (a *AppService) handleTransaction(w http.ResponseWriter, r *http.Request) {
	if !a.authOK(r) {
		a.denyUnauthed(w, r)
		return
	}
	txnID := r.PathValue("txnId")
	if txnID == "" {
		writeError(w, http.StatusBadRequest, "M_BAD_JSON", "missing txnId")
		return
	}

	// Idempotency (spec): a retried, already-processed transaction is a no-op.
	if done, err := a.store.HasTxn(txnID); err != nil {
		a.log.Error("txn dedup read failed", "txn", txnID, "err", err)
	} else if done {
		writeJSON(w, http.StatusOK, struct{}{})
		return
	}

	var txn struct {
		Events []Event `json:"events"`
	}
	if err := json.NewDecoder(r.Body).Decode(&txn); err != nil {
		writeError(w, http.StatusBadRequest, "M_NOT_JSON", "invalid transaction body")
		return
	}
	a.log.Debug("transaction received", "txn", txnID, "events", len(txn.Events))

	// Process with the bot's long-lived context (not the request context) so a
	// homeserver-side timeout can't cancel an in-flight reply mid-send.
	a.handler(a.baseCtx, txn.Events)

	if err := a.store.MarkTxn(txnID); err != nil {
		a.log.Error("txn mark failed", "txn", txnID, "err", err)
	}
	writeJSON(w, http.StatusOK, struct{}{})
}

func (a *AppService) handleUserQuery(w http.ResponseWriter, r *http.Request) {
	if !a.authOK(r) {
		a.denyUnauthed(w, r)
		return
	}
	// We own exactly one user. Synapse auto-creates the sender_localpart user;
	// confirm it for our mxid, 404 for anything else in (an over-broad) namespace.
	if r.PathValue("userId") == a.cfg.BotMXID {
		writeJSON(w, http.StatusOK, struct{}{})
		return
	}
	writeError(w, http.StatusNotFound, "M_NOT_FOUND", "no such user")
}

func (a *AppService) handleRoomQuery(w http.ResponseWriter, r *http.Request) {
	if !a.authOK(r) {
		a.denyUnauthed(w, r)
		return
	}
	// The bot claims no room aliases.
	writeError(w, http.StatusNotFound, "M_NOT_FOUND", "no such room")
}

func writeJSON(w http.ResponseWriter, status int, body any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(body)
}

func writeError(w http.ResponseWriter, status int, code, msg string) {
	writeJSON(w, status, map[string]string{"errcode": code, "error": msg})
}

// denyUnauthed logs and rejects a request whose hs_token didn't match. Logging
// at WARN makes probing / a misconfigured homeserver visible (the token itself
// is never logged).
func (a *AppService) denyUnauthed(w http.ResponseWriter, r *http.Request) {
	a.log.Warn("rejected request: bad hs_token", "method", r.Method, "path", r.URL.Path, "remote", r.RemoteAddr)
	writeError(w, http.StatusForbidden, "M_FORBIDDEN", "bad hs_token")
}