golib/cmd/smsapid/main.go

package main

import (
	"context"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"

	chiMiddleware "github.com/go-chi/chi/v5/middleware"
	"github.com/jszwec/csvutil"
	"github.com/simonfrey/jsonl"
	"github.com/therootcompany/golib/auth"
	"github.com/therootcompany/golib/auth/csvauth"
	"github.com/therootcompany/golib/colorjson"
	"github.com/therootcompany/golib/http/androidsmsgateway"
	mwpkg "github.com/therootcompany/golib/http/middleware/v2"
)

var jsonf = colorjson.NewFormatter()

// webhookMux protects webhookWriter, pingWriter, webhookEvents, and pingEvents.
var webhookMux = sync.Mutex{}
var webhookEvents []androidsmsgateway.WebhookEvent
var webhookWriter jsonl.Writer
var pingEvents []*androidsmsgateway.WebhookPing
var pingWriter jsonl.Writer

var smsAuth *csvauth.Auth

var smsRequestAuth = &auth.RequestAuthenticator{
	AuthenticateBasic:    true,
	AuthorizationSchemes: []string{"*"},
	TokenHeaders:         []string{"API-Key", "X-API-Key"},
}

func main() {
	jsonf.Indent = 3

	messagesPath := "./messages.jsonl"
	{
		file, err := os.OpenFile(messagesPath, os.O_RDONLY|os.O_CREATE, 0644)
		if err != nil {
			log.Fatalf("failed to open file '%s': %v", messagesPath, err)
		}
		defer func() { _ = file.Close() }()

		webhookEvents, err = readWebhooks(file)
		if err != nil {
			log.Fatalf("failed to read jsonl file '%s': %v", messagesPath, err)
		}
	}
	{
		file, err := os.OpenFile(messagesPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
		if err != nil {
			panic(fmt.Errorf("failed to open file: %v", err))
		}
		defer func() { _ = file.Close() }()

		webhookWriter = jsonl.NewWriter(file)
	}

	pingsPath := "./pings.jsonl"
	{
		file, err := os.OpenFile(pingsPath, os.O_RDONLY|os.O_CREATE, 0644)
		if err != nil {
			log.Fatalf("failed to open file '%s': %v", pingsPath, err)
		}
		defer func() { _ = file.Close() }()

		pingEvents, err = readPings(file)
		if err != nil {
			log.Fatalf("failed to read jsonl file '%s': %v", pingsPath, err)
		}
	}
	{
		file, err := os.OpenFile(pingsPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
		if err != nil {
			panic(fmt.Errorf("failed to open file: %v", err))
		}
		defer func() { _ = file.Close() }()

		pingWriter = jsonl.NewWriter(file)
	}

	// Load credentials for /api/smsgw routes.
	credPath := "./credentials.tsv"
	if v := os.Getenv("SMSAPID_CREDENTIALS_FILE"); v != "" {
		credPath = v
	}
	aesKey := getAESKey("CSVAUTH_AES_128_KEY")
	if f, err := os.Open(credPath); err == nil {
		defer func() { _ = f.Close() }()
		smsAuth = csvauth.New(aesKey)
		if err := smsAuth.LoadCSV(f, '\t'); err != nil {
			log.Fatalf("failed to load credentials from %q: %v", credPath, err)
		}
		smsRequestAuth.Authenticator = smsAuth
	} else {
		log.Printf("Warning: credentials file %q not found; /api/smsgw routes will return 401: %v", credPath, err)
	}

	mux := http.NewServeMux()
	mux.HandleFunc("GET /api/webhooks", handlerWebhooks)
	mux.Handle("GET /", LogRequest(http.HandlerFunc(HandleOK)))
	mux.Handle("POST /", LogRequest(http.HandlerFunc(handler)))
	mux.Handle("PATCH /", LogRequest(http.HandlerFunc(HandleOK)))
	mux.Handle("PUT /", LogRequest(http.HandlerFunc(HandleOK)))
	mux.Handle("DELETE /", LogRequest(http.HandlerFunc(HandleOK)))

	// Protected routes under /api/smsgw, each guarded by its specific sms:* permission.
	smsgw := mwpkg.WithMux(mux)
	smsgw.With(requireSMSPermission("sms:received")).HandleFunc("GET /api/smsgw/received.csv", handlerReceivedCSV)
	smsgw.With(requireSMSPermission("sms:received")).HandleFunc("GET /api/smsgw/received.json", handlerReceivedJSON)
	smsgw.With(requireSMSPermission("sms:sent")).HandleFunc("GET /api/smsgw/sent.csv", handlerSentCSV)
	smsgw.With(requireSMSPermission("sms:sent")).HandleFunc("GET /api/smsgw/sent.json", handlerSentJSON)
	smsgw.With(requireSMSPermission("sms:ping")).HandleFunc("GET /api/smsgw/ping.csv", handlerPingCSV)
	smsgw.With(requireSMSPermission("sms:ping")).HandleFunc("GET /api/smsgw/ping.json", handlerPingJSON)

	addr := "localhost:8088"
	fmt.Printf("Listening on %s...\n\n", addr)
	log.Fatal(http.ListenAndServe(addr, chiMiddleware.Logger(chiMiddleware.Compress(5)(mux))))
}

// hasSMSPermission reports whether perms includes the wildcard "sms:*" or the specific permission.
func hasSMSPermission(perms []string, permission string) bool {
	for _, p := range perms {
		if p == "sms:*" || p == permission {
			return true
		}
	}
	return false
}

// requireSMSPermission returns a middleware that authenticates the request and enforces
// that the credential holds "sms:*" or the given specific permission.
func requireSMSPermission(permission string) func(http.Handler) http.Handler {
	return func(next http.Handler) http.Handler {
		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			if smsAuth == nil {
				w.Header().Set("WWW-Authenticate", `Basic`)
				http.Error(w, "Unauthorized", http.StatusUnauthorized)
				return
			}
			cred, err := smsRequestAuth.Authenticate(r)
			if err != nil || !hasSMSPermission(cred.Permissions(), permission) {
				w.Header().Set("WWW-Authenticate", `Basic`)
				http.Error(w, "Unauthorized", http.StatusUnauthorized)
				return
			}
			next.ServeHTTP(w, r)
		})
	}
}

// getAESKey reads an AES-128 key (32 hex chars) from an environment variable.
// Returns a zero key if the variable is absent or invalid.
func getAESKey(envname string) []byte {
	if v := os.Getenv(envname); v != "" {
		key, err := hex.DecodeString(strings.TrimSpace(v))
		if err == nil && len(key) == 16 {
			return key
		}
		log.Printf("Warning: %s is invalid (must be 32 hex chars); using zero key", envname)
	}
	return make([]byte, 16)
}

// parseSinceLimit extracts the "since" (ISO datetime) and "limit" query parameters.
func parseSinceLimit(r *http.Request) (time.Time, int) {
	var since time.Time
	if s := r.URL.Query().Get("since"); s != "" {
		for _, format := range []string{time.RFC3339, "2006-01-02T15:04:05-0700", "2006-01-02"} {
			if t, err := time.Parse(format, s); err == nil {
				since = t
				break
			}
		}
	}

	limit := 10000
	if l := r.URL.Query().Get("limit"); l != "" {
		if n, err := strconv.Atoi(strings.ReplaceAll(l, "_", "")); err == nil && n > 0 {
			limit = n
		}
	}

	return since, limit
}

func handlerReceivedCSV(w http.ResponseWriter, r *http.Request) {
	since, limit := parseSinceLimit(r)

	webhookMux.Lock()
	rows := make([]*androidsmsgateway.WebhookReceived, 0, min(len(webhookEvents), limit))
	for _, event := range webhookEvents {
		recv, ok := event.(*androidsmsgateway.WebhookReceived)
		if !ok {
			continue
		}
		if !since.IsZero() && !recv.Payload.ReceivedAt.After(since) {
			continue
		}
		rows = append(rows, recv)
		if len(rows) >= limit {
			break
		}
	}
	webhookMux.Unlock()

	b, err := csvutil.Marshal(rows)
	if err != nil {
		http.Error(w, `{"error":"failed to encode CSV"}`, http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "text/csv; charset=utf-8")
	_, _ = w.Write(b)
}

func handlerReceivedJSON(w http.ResponseWriter, r *http.Request) {
	since, limit := parseSinceLimit(r)

	webhookMux.Lock()
	rows := make([]*androidsmsgateway.WebhookReceived, 0, min(len(webhookEvents), limit))
	for _, event := range webhookEvents {
		recv, ok := event.(*androidsmsgateway.WebhookReceived)
		if !ok {
			continue
		}
		if !since.IsZero() && !recv.Payload.ReceivedAt.After(since) {
			continue
		}
		rows = append(rows, recv)
		if len(rows) >= limit {
			break
		}
	}
	webhookMux.Unlock()

	w.Header().Set("Content-Type", "application/json")
	enc := json.NewEncoder(w)
	enc.SetEscapeHTML(false)
	_ = enc.Encode(rows)
}

func handlerSentCSV(w http.ResponseWriter, r *http.Request) {
	since, limit := parseSinceLimit(r)

	webhookMux.Lock()
	rows := make([]*androidsmsgateway.WebhookSent, 0, min(len(webhookEvents), limit))
	for _, event := range webhookEvents {
		sent, ok := event.(*androidsmsgateway.WebhookSent)
		if !ok {
			continue
		}
		if !since.IsZero() && !sent.Payload.SentAt.After(since) {
			continue
		}
		rows = append(rows, sent)
		if len(rows) >= limit {
			break
		}
	}
	webhookMux.Unlock()

	b, err := csvutil.Marshal(rows)
	if err != nil {
		http.Error(w, `{"error":"failed to encode CSV"}`, http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "text/csv; charset=utf-8")
	_, _ = w.Write(b)
}

func handlerSentJSON(w http.ResponseWriter, r *http.Request) {
	since, limit := parseSinceLimit(r)

	webhookMux.Lock()
	rows := make([]*androidsmsgateway.WebhookSent, 0, min(len(webhookEvents), limit))
	for _, event := range webhookEvents {
		sent, ok := event.(*androidsmsgateway.WebhookSent)
		if !ok {
			continue
		}
		if !since.IsZero() && !sent.Payload.SentAt.After(since) {
			continue
		}
		rows = append(rows, sent)
		if len(rows) >= limit {
			break
		}
	}
	webhookMux.Unlock()

	w.Header().Set("Content-Type", "application/json")
	enc := json.NewEncoder(w)
	enc.SetEscapeHTML(false)
	_ = enc.Encode(rows)
}

func handlerPingCSV(w http.ResponseWriter, r *http.Request) {
	since, limit := parseSinceLimit(r)

	webhookMux.Lock()
	rows := make([]*androidsmsgateway.WebhookPing, 0, min(len(pingEvents), limit))
	for _, ping := range pingEvents {
		pingedAt := ping.PingedAt
		if pingedAt.IsZero() {
			pingedAt = time.UnixMilli(ping.XTimestamp).UTC()
		}
		if !since.IsZero() && !pingedAt.After(since) {
			continue
		}
		rows = append(rows, ping)
		if len(rows) >= limit {
			break
		}
	}
	webhookMux.Unlock()

	b, err := csvutil.Marshal(rows)
	if err != nil {
		http.Error(w, `{"error":"failed to encode CSV"}`, http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "text/csv; charset=utf-8")
	_, _ = w.Write(b)
}

func handlerPingJSON(w http.ResponseWriter, r *http.Request) {
	since, limit := parseSinceLimit(r)

	webhookMux.Lock()
	rows := make([]*androidsmsgateway.WebhookPing, 0, min(len(pingEvents), limit))
	for _, ping := range pingEvents {
		pingedAt := ping.PingedAt
		if pingedAt.IsZero() {
			pingedAt = time.UnixMilli(ping.XTimestamp).UTC()
		}
		if !since.IsZero() && !pingedAt.After(since) {
			continue
		}
		rows = append(rows, ping)
		if len(rows) >= limit {
			break
		}
	}
	webhookMux.Unlock()

	w.Header().Set("Content-Type", "application/json")
	enc := json.NewEncoder(w)
	enc.SetEscapeHTML(false)
	_ = enc.Encode(rows)
}

func HandleOK(w http.ResponseWriter, r *http.Request) {
	w.WriteHeader(http.StatusOK)
}

type ctxKey struct{}

var CtxKeyBody = ctxKey{}

func LogRequest(next http.Handler) http.Handler {
	return LogHeaders(LogBody(next))
}

func LogHeaders(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		// Log method, path, and query
		var query string
		if len(r.URL.RawQuery) > 0 {
			query = "?" + r.URL.RawQuery
		}
		log.Printf("%s %s%s", r.Method, r.URL.Path, query)

		// Find max header name length for alignment
		maxLen := len("HOST")
		for name := range r.Header {
			if len(name) > maxLen {
				maxLen = len(name)
			}
		}
		maxLen += 1

		fmt.Printf("   %-"+fmt.Sprintf("%d", maxLen+1)+"s %s\n", "HOST", r.Host)
		for name, values := range r.Header {
			for _, value := range values {
				fmt.Printf("   %-"+fmt.Sprintf("%d", maxLen+1)+"s %s\n", name+":", value)
			}
		}
		fmt.Fprintf(os.Stderr, "\n")

		next.ServeHTTP(w, r)
	})
}

func LogBody(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		body, err := io.ReadAll(r.Body)
		switch strings.ToUpper(r.Method) {
		case "HEAD", "GET", "DELETE", "OPTIONS":
			if len(body) > 0 {
				fmt.Fprintf(os.Stderr, "Unexpected body:\n%q\n", string(body))
			}
		case "POST", "PATCH", "PUT":
		// known
		default:
			fmt.Fprintf(os.Stderr, "Unexpected method %s\n", r.Method)
		}
		defer fmt.Println()

		if err != nil {
			fmt.Fprintf(os.Stderr, "Failed to read body:\n%q\n", string(body))
			return
		}

		// Parse and pretty-print JSON, or raw body
		textBytes := body
		var text string
		var data any
		if err := json.Unmarshal(body, &data); err == nil {
			textBytes, _ = jsonf.Marshal(data)
		}
		text = string(textBytes)
		text = prefixLines(text, "   ")
		text = strings.TrimSpace(text)
		fmt.Printf("   %s\n", text)

		ctx := context.WithValue(r.Context(), CtxKeyBody, body)
		next.ServeHTTP(w, r.WithContext(ctx))
	})
}

func handlerWebhooks(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "application/json")
	enc := json.NewEncoder(w)
	enc.SetEscapeHTML(false)

	next := r.URL.Query().Get("next")
	previous := r.URL.Query().Get("previous")
	limitStr := r.URL.Query().Get("limit")
	limit, err := strconv.Atoi(limitStr)
	if err != nil || limit <= 0 {
		limit = 1000
	}

	var startIdx, endIdx int
	if next != "" {
		for i, event := range webhookEvents {
			switch e := event.(type) {
			case *androidsmsgateway.WebhookSent:
				if e.ID == next {
					startIdx = i + 1
					break
				}
			case *androidsmsgateway.WebhookDelivered:
				if e.ID == next {
					startIdx = i + 1
					break
				}
			case *androidsmsgateway.WebhookReceived:
				if e.ID == next {
					startIdx = i + 1
					break
				}
			}
		}
	} else if previous != "" {
		for i, event := range webhookEvents {
			switch e := event.(type) {
			case *androidsmsgateway.WebhookSent:
				if e.ID == previous && i >= limit {
					startIdx = i - limit
					break
				}
			case *androidsmsgateway.WebhookDelivered:
				if e.ID == previous && i >= limit {
					startIdx = i - limit
					break
				}
			case *androidsmsgateway.WebhookReceived:
				if e.ID == previous && i >= limit {
					startIdx = i - limit
					break
				}
			}
		}
	} else {
		if len(webhookEvents) > limit {
			startIdx = len(webhookEvents) - limit
		} else {
			startIdx = 0
		}
	}

	endIdx = min(startIdx+limit, len(webhookEvents))

	if _, err := w.Write([]byte("[")); err != nil {
		http.Error(w, `{"error":"failed to write response"}`, http.StatusInternalServerError)
		return
	}
	for i, event := range webhookEvents[startIdx:endIdx] {
		if i > 0 {
			if _, err := w.Write([]byte(",")); err != nil {
				http.Error(w, `{"error":"failed to write response"}`, http.StatusInternalServerError)
				return
			}
		}
		if err := enc.Encode(event); err != nil {
			http.Error(w, `{"error":"failed to encode webhook"}`, http.StatusInternalServerError)
			return
		}
	}
	if _, err := w.Write([]byte("]")); err != nil {
		http.Error(w, `{"error":"failed to write response"}`, http.StatusInternalServerError)
		return
	}
}

func handler(w http.ResponseWriter, r *http.Request) {
	// this will return OK unless a retry is needed (e.g. internal error)

	body, ok := r.Context().Value(CtxKeyBody).([]byte)
	if !ok {
		return
	}

	var webhook androidsmsgateway.Webhook
	if err := json.Unmarshal(body, &webhook); err != nil {
		http.Error(w, `{"error":"failed to parse webhook"}`, http.StatusOK)
		return
	}
	ts, _ := strconv.Atoi(r.Header.Get("X-Timestamp"))
	webhook.XTimestamp = int64(ts)
	webhook.XSignature = r.Header.Get("X-Signature")

	h, err := androidsmsgateway.Decode(&webhook)
	if err != nil {
		http.Error(w, `{"error":"failed to parse webhook as a specific event"}`, http.StatusOK)
		return
	}

	switch h.GetEvent() {
	case "system:ping":
		ping := h.(*androidsmsgateway.WebhookPing)
		ping.PingedAt = time.UnixMilli(webhook.XTimestamp).UTC()
		webhookMux.Lock()
		defer webhookMux.Unlock()
		if err := pingWriter.Write(ping); err != nil {
			http.Error(w, `{"error":"failed to save ping"}`, http.StatusOK)
			return
		}
		pingEvents = append(pingEvents, ping)
	case "mms:received", "sms:received", "sms:data-received", "sms:sent", "sms:delivered", "sms:failed":
		webhookMux.Lock()
		defer webhookMux.Unlock()
		if err := webhookWriter.Write(h); err != nil {
			http.Error(w, `{"error":"failed to save webhook"}`, http.StatusOK)
			return
		}
		webhookEvents = append(webhookEvents, h)
	default:
		http.Error(w, `{"error":"unknown webhook event"}`, http.StatusOK)
		return
	}

	_, _ = w.Write([]byte(`{"message": "ok"}`))
}

func prefixLines(text, prefix string) string {
	lines := strings.Split(text, "\n")
	for i, line := range lines {
		lines[i] = prefix + line
	}
	return strings.Join(lines, "\n")
}

func readWebhooks(f io.Reader) ([]androidsmsgateway.WebhookEvent, error) {
	var webhooks []androidsmsgateway.WebhookEvent
	r := jsonl.NewReader(f)
	err := r.ReadLines(func(line []byte) error {
		if len(line) == 0 {
			return nil
		}

		var webhook androidsmsgateway.Webhook
		if err := json.Unmarshal(line, &webhook); err != nil {
			return fmt.Errorf("could not unmarshal into Webhook: %w", err)
		}

		switch webhook.Event {
		case "sms:sent":
			var sent androidsmsgateway.WebhookSent
			if err := json.Unmarshal(line, &sent); err != nil {
				return fmt.Errorf("could not unmarshal into WebhookSent: %w", err)
			}
			webhooks = append(webhooks, &sent)
		case "sms:delivered":
			var delivered androidsmsgateway.WebhookDelivered
			if err := json.Unmarshal(line, &delivered); err != nil {
				return fmt.Errorf("could not unmarshal into WebhookDelivered: %w", err)
			}
			webhooks = append(webhooks, &delivered)
		case "sms:received":
			var received androidsmsgateway.WebhookReceived
			if err := json.Unmarshal(line, &received); err != nil {
				return fmt.Errorf("could not unmarshal into WebhookReceived: %w", err)
			}
			webhooks = append(webhooks, &received)
		default:
			return fmt.Errorf("unknown event type: %s", webhook.Event)
		}
		return nil
	})

	if err != nil {
		return webhooks, fmt.Errorf("failed to read JSONL lines: %w", err)
	}
	return webhooks, nil
}

func readPings(f io.Reader) ([]*androidsmsgateway.WebhookPing, error) {
	var pings []*androidsmsgateway.WebhookPing
	r := jsonl.NewReader(f)
	err := r.ReadLines(func(line []byte) error {
		if len(line) == 0 {
			return nil
		}
		var ping androidsmsgateway.WebhookPing
		if err := json.Unmarshal(line, &ping); err != nil {
			return fmt.Errorf("could not unmarshal into WebhookPing: %w", err)
		}
		pings = append(pings, &ping)
		return nil
	})

	if err != nil {
		return pings, fmt.Errorf("failed to read JSONL lines: %w", err)
	}
	return pings, nil
}