memby/server/internal/integrations/integrations.go

// Package integrations delivers administrative events to external services.
//
// It is a *subscriber* on the event bus, not a call made from the code that produces the
// events. handleLogin knows nothing about Discord; it publishes a sign-in, and whether
// that becomes a message in a channel is decided here, from the operator's configuration,
// on a goroutine of its own. That is the whole design: adding a second destination is a
// new Transport in this package and nothing else in the gateway changes.
package integrations

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"strings"
	"sync"
	"time"

	"github.com/ponzischeme89/memby/server/internal/adminevents"
	"github.com/ponzischeme89/memby/server/internal/notify"
	"github.com/ponzischeme89/memby/server/internal/store"
)

// queueDepth bounds the work waiting to be delivered.
//
// Deliberately lossy at the limit. An outbound webhook is a network call to somebody
// else's service, and the worst failure mode available here is one slow endpoint holding
// up the gateway — so a full queue drops the event and says so, rather than applying back
// pressure to a television signing in.
const queueDepth = 256

// settingsTTL is how long a configuration read is reused. Every event would otherwise be
// a Postgres read on the delivery path, and an operator's change reaching the dispatcher
// within a few seconds is soon enough for something whose own delivery is best-effort.
const settingsTTL = 15 * time.Second

// requestTimeout bounds one delivery.
const requestTimeout = 10 * time.Second

// Transport is one kind of destination. A new integration implements this and registers
// itself in transports below; nothing else in the package has a case per kind.
type Transport interface {
	// Kind is the stored discriminator, e.g. "discord".
	Kind() string
	// Deliver posts the event and reports the HTTP status it got, where there was one.
	Deliver(ctx context.Context, client *http.Client, integration store.Integration, event adminevents.Event) (int, error)
}

type job struct {
	event adminevents.Event
	// only is set by a test delivery: one destination, regardless of its event
	// selection, because the operator has just pressed Test on that row.
	only string
}

// Dispatcher is the bus sink. One instance, created at start-up and registered with
// [adminevents.Bus.AddSink].
type Dispatcher struct {
	store  *store.Store
	log    *slog.Logger
	client *http.Client
	events *adminevents.Bus
	// notify is the audit trail every outbound notification lands in. This package is the
	// one producer that reports to it rather than being driven by it: the dispatcher has
	// its own queue, pacing and transport registry, and routing deliveries through
	// notify.Send would make the audit trail the thing deciding what Discord receives.
	notify *notify.Service

	transports map[string]Transport
	queue      chan job
	paused     func() bool

	mu       sync.Mutex
	cached   store.IntegrationSettings
	cachedAt time.Time

	// dropped counts events the queue could not take. It is reported on the console's
	// integrations page: a dispatcher quietly losing events is otherwise indistinguishable
	// from a household in which nothing has happened.
	dropped int64
}

// SetPaused installs the server-wide quiet-time gate before Start is called.
func (d *Dispatcher) SetPaused(paused func() bool) { d.paused = paused }

func New(
	st *store.Store, log *slog.Logger, events *adminevents.Bus, notifier *notify.Service,
) *Dispatcher {
	dispatcher := &Dispatcher{
		store: st, log: log.With("component", "integrations"), events: events,
		notify:     notifier,
		client:     &http.Client{Timeout: requestTimeout},
		transports: map[string]Transport{},
		queue:      make(chan job, queueDepth),
	}
	dispatcher.register(discordTransport{})
	return dispatcher
}

func (d *Dispatcher) register(transport Transport) {
	d.transports[transport.Kind()] = transport
}

// Start begins the delivery worker. One worker, not a pool: household volume is a handful
// of events a minute, and serialising them means a destination cannot be hit with four
// concurrent posts by a burst.
func (d *Dispatcher) Start(ctx context.Context) {
	go func() {
		for {
			select {
			case <-ctx.Done():
				return
			case work := <-d.queue:
				for d.paused != nil && d.paused() {
					timer := time.NewTimer(30 * time.Second)
					select {
					case <-ctx.Done():
						timer.Stop()
						return
					case <-timer.C:
					}
				}
				d.deliver(ctx, work)
			}
		}
	}()
}

// DeliverAdminEvent implements adminevents.Sink. It only enqueues: the bus calls this
// while a sign-in is in progress, so nothing here may touch a network.
func (d *Dispatcher) DeliverAdminEvent(_ context.Context, event adminevents.Event) {
	select {
	case d.queue <- job{event: event}:
	default:
		d.mu.Lock()
		d.dropped++
		dropped := d.dropped
		d.mu.Unlock()
		d.log.Warn("integration event dropped", "type", event.Type, "dropped_total", dropped)
	}
}

// Test delivers a synthetic event to one destination and reports what happened, so the
// operator gets the answer in the response to the button they pressed rather than having
// to go and read a delivery history.
//
// It is deliberately not queued: a test is a question, and an answer that arrives some
// time later on another page is not one.
func (d *Dispatcher) Test(ctx context.Context, integrationID string) error {
	settings, err := d.settings(ctx, true)
	if err != nil {
		return err
	}
	for _, integration := range settings.Integrations {
		if integration.ID != integrationID {
			continue
		}
		if strings.TrimSpace(integration.URL) == "" {
			return errors.New("this integration has no webhook address")
		}
		event := adminevents.Event{
			OccurredAt: time.Now(),
			Type:       "integration.test",
			Severity:   adminevents.SeverityInfo,
			Title:      "Test message from Memby",
			Summary:    "If you can read this, " + integration.Name + " is configured correctly.",
			Actor:      "admin console",
			Target:     integration.Name,
		}
		return d.post(ctx, integration, event)
	}
	return errors.New("no such integration")
}

func (d *Dispatcher) deliver(ctx context.Context, work job) {
	settings, err := d.settings(ctx, false)
	if err != nil {
		d.log.Warn("integration settings unavailable", "error", err)
		return
	}
	for _, integration := range settings.Integrations {
		if work.only != "" && integration.ID != work.only {
			continue
		}
		if !integration.Enabled || !wants(integration, work.event.Type) {
			continue
		}
		if err := d.post(ctx, integration, work.event); err != nil {
			d.announceFailure(ctx, integration, work.event, err)
		}
	}
}

// wants is the event selection rule, kept apart from the loop because it is the one thing
// here worth being sure about: a destination sends what it was ticked for and nothing
// else. An empty selection sends nothing — see store.Integration.
func wants(integration store.Integration, eventType string) bool {
	for _, selected := range integration.Events {
		if selected == eventType {
			return true
		}
	}
	return false
}

func (d *Dispatcher) post(ctx context.Context, integration store.Integration, event adminevents.Event) error {
	transport, ok := d.transports[integration.Kind]
	if !ok {
		return fmt.Errorf("no transport for %q", integration.Kind)
	}
	started := time.Now()
	postCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), requestTimeout)
	status, err := transport.Deliver(postCtx, d.client, integration, event)
	cancel()

	message := ""
	if err != nil {
		message = err.Error()
	}
	took := time.Since(started)
	if d.store != nil {
		record := store.IntegrationDelivery{
			IntegrationID: integration.ID, EventType: event.Type,
			Success: err == nil, StatusCode: status,
			DurationMS: took.Milliseconds(), Error: message,
		}
		if writeErr := d.store.RecordIntegrationDelivery(
			context.WithoutCancel(ctx), record,
		); writeErr != nil {
			d.log.Warn("delivery not recorded", "integration", integration.ID, "error", writeErr)
		}
	}
	// The per-integration delivery history above answers "is this destination healthy",
	// which is what the integrations page asks. This is the other question — "did Memby
	// tell anybody about that event" — and it is answered in one place for every channel,
	// which is the whole reason the notification log exists.
	d.notify.Log(ctx, notify.Notification{
		Channel: notify.ChannelWebhook,
		Kind:    event.Type,
		Source:  "integrations",
		Title:   event.Title,
		Body:    event.Summary,
		// The destination's NAME, never its address: a Discord webhook URL is the
		// credential, and this row is rendered in the console.
		Target:    integration.Name,
		SourceKey: integration.ID,
		Metadata: map[string]any{
			"integrationId": integration.ID,
			"kind":          integration.Kind,
			"statusCode":    status,
		},
	}, deliveryOutcome(status, err), took)
	return err
}

// deliveryOutcome turns a transport's answer into an audit status.
//
// A webhook is the one channel that gets Delivered rather than Sent: somebody else's
// service actually acknowledged this, where writing a row into a viewer's list is
// finished the moment it returns with nobody to confirm it. The status code is kept in
// the detail because "failed" on its own sends an operator to the wrong place — a 404 is
// a webhook that has been deleted, a 429 is one that is merely busy.
func deliveryOutcome(status int, err error) notify.Outcome {
	if err != nil {
		if status > 0 {
			return notify.Outcome{
				Status: notify.StatusFailed,
				Detail: fmt.Sprintf("HTTP %d: %s", status, err.Error()),
				Err:    err,
			}
		}
		return notify.Failed(err)
	}
	if status > 0 {
		return notify.Delivered(fmt.Sprintf("HTTP %d", status))
	}
	return notify.Delivered("")
}

// announceFailure puts a failed delivery back into the feed the operator is reading.
//
// It publishes a *different* type from the event that failed, and integration.failed is
// the one type a destination cannot usefully be subscribed to — a webhook that is down
// cannot be told that it is down, and a second destination being told about the first is
// the case this exists to serve.
func (d *Dispatcher) announceFailure(
	ctx context.Context, integration store.Integration,
	event adminevents.Event, err error,
) {
	d.log.Warn("integration delivery failed",
		"integration", integration.ID, "name", integration.Name,
		"event", event.Type, "error", err)
	if event.Type == adminevents.TypeIntegrationFailed {
		// A failure announcing a failure would loop for as long as the endpoint is down.
		return
	}
	d.events.Publish(ctx, adminevents.Event{
		Type:     adminevents.TypeIntegrationFailed,
		Severity: adminevents.SeverityWarning,
		Title:    integration.Name + " could not be reached",
		Summary:  err.Error(),
		Actor:    "integrations",
		Target:   integration.Name,
		Link:     "/admin/integrations",
		Metadata: adminevents.Meta(map[string]any{
			"integrationId": integration.ID, "eventType": event.Type,
		}),
	})
}

func (d *Dispatcher) settings(ctx context.Context, fresh bool) (store.IntegrationSettings, error) {
	d.mu.Lock()
	if !fresh && time.Since(d.cachedAt) < settingsTTL {
		cached := d.cached
		d.mu.Unlock()
		return cached, nil
	}
	d.mu.Unlock()

	if d.store == nil {
		return store.IntegrationSettings{}, nil
	}
	settings, err := d.store.Integrations(ctx)
	if err != nil {
		return store.IntegrationSettings{}, err
	}
	d.mu.Lock()
	d.cached, d.cachedAt = settings, time.Now()
	d.mu.Unlock()
	return settings, nil
}

// Invalidate drops the cached configuration, so an operator's save takes effect on the
// next event rather than up to settingsTTL later. Called from the admin handler that
// writes it.
func (d *Dispatcher) Invalidate() {
	if d == nil {
		return
	}
	d.mu.Lock()
	d.cachedAt = time.Time{}
	d.mu.Unlock()
}

// Dropped reports how many events the queue could not take.
func (d *Dispatcher) Dropped() int64 {
	if d == nil {
		return 0
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	return d.dropped
}

// --- Discord ----------------------------------------------------------------------

type discordTransport struct{}

func (discordTransport) Kind() string { return store.IntegrationDiscord }

// discordEmbed is the subset of Discord's webhook payload this uses. An embed rather than
// plain content because the colour down its left edge is the one thing that makes a
// failure distinguishable from an ordinary sign-in at a glance in a busy channel.
type discordEmbed struct {
	Title       string              `json:"title,omitempty"`
	Description string              `json:"description,omitempty"`
	Color       int                 `json:"color,omitempty"`
	Timestamp   string              `json:"timestamp,omitempty"`
	Fields      []discordEmbedField `json:"fields,omitempty"`
	Footer      *discordFooter      `json:"footer,omitempty"`
}

type discordEmbedField struct {
	Name   string `json:"name"`
	Value  string `json:"value"`
	Inline bool   `json:"inline"`
}

type discordFooter struct {
	Text string `json:"text"`
}

type discordPayload struct {
	Username string         `json:"username,omitempty"`
	Embeds   []discordEmbed `json:"embeds"`
}

// severityColour matches the console's own tones: green is a verdict, amber is look at
// this, red is wrong. Same three colours in the channel as on the page, so an operator
// reading one is not learning a second vocabulary.
func severityColour(severity string) int {
	switch severity {
	case adminevents.SeverityError:
		return 0xE5484D
	case adminevents.SeverityWarning:
		return 0xF5A524
	default:
		return 0x3FB950
	}
}

func (discordTransport) Deliver(
	ctx context.Context, client *http.Client,
	integration store.Integration, event adminevents.Event,
) (int, error) {
	embed := discordEmbed{
		Title:       strings.TrimSpace(event.Title),
		Description: strings.TrimSpace(event.Summary),
		Color:       severityColour(event.Severity),
		Timestamp:   event.OccurredAt.UTC().Format(time.RFC3339),
		Footer:      &discordFooter{Text: "Memby · " + event.Type},
	}
	if embed.Title == "" {
		embed.Title = event.Type
	}
	if event.Actor != "" {
		embed.Fields = append(embed.Fields,
			discordEmbedField{Name: "Actor", Value: event.Actor, Inline: true})
	}
	if event.Target != "" {
		embed.Fields = append(embed.Fields,
			discordEmbedField{Name: "Target", Value: event.Target, Inline: true})
	}

	body, err := json.Marshal(discordPayload{
		Username: "Memby", Embeds: []discordEmbed{embed},
	})
	if err != nil {
		return 0, err
	}
	request, err := http.NewRequestWithContext(
		ctx, http.MethodPost, integration.URL, bytes.NewReader(body))
	if err != nil {
		return 0, err
	}
	request.Header.Set("Content-Type", "application/json")

	response, err := client.Do(request)
	if err != nil {
		// Never include the request URL in the error: it is the webhook token, and this
		// string is stored in the delivery history and shown in the console.
		return 0, errors.New("could not reach the webhook")
	}
	defer response.Body.Close()
	// Discord answers 204 with no body on success. Anything else is reported by status
	// alone — its error bodies are not worth surfacing to an operator and a 401 body has
	// been known to echo request details back.
	if response.StatusCode < 200 || response.StatusCode >= 300 {
		return response.StatusCode, fmt.Errorf("webhook returned %d", response.StatusCode)
	}
	return response.StatusCode, nil
}