beacon-server/cmd/tower/main.go

package main

import (
	"bytes"
	"context"
	"encoding/hex"
	"fmt"
	"log"
	"net/http"
	"os"
	"os/signal"
	"syscall"

	"github.com/MeshCore-Tower/tower-server/db"
	"github.com/MeshCore-Tower/tower-server/internal/api/router"
	"github.com/MeshCore-Tower/tower-server/internal/config"
	"github.com/MeshCore-Tower/tower-server/internal/hub"
	"github.com/MeshCore-Tower/tower-server/internal/ingest"
	"github.com/MeshCore-Tower/tower-server/internal/keystore"

	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/joho/godotenv"
)

func main() {
	_ = godotenv.Load()
	addr := os.Getenv("LISTEN_ADDR")
	if addr == "" {
		addr = ":8080"
	}

	configPath := os.Getenv("CONFIG_PATH")
	if configPath == "" {
		configPath = "config.yaml"
	}

	cfg, err := config.Load(configPath)
	if err != nil {
		log.Fatalf("failed to load config: %v", err)
	}
	// ── Hub ──────────────────────────────────────────────────────────────────
	h := hub.New()
	go h.Run()

	// ── MQTT ingest workers ──────────────────────────────────────────────────
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	pool, err := pgxpool.New(ctx, mustEnv("POSTGRES_DSN"))
	if err != nil {
		log.Fatalf("failed to connect to postgres at %s: %v", os.Getenv("POSTGRES_DSN_HOST"), err)
	}
	defer pool.Close()

	store := db.New(pool)

	// ── Seed config data ─────────────────────────────────────────────────────
	if err := config.Seed(ctx, cfg, store); err != nil {
		log.Fatalf("failed to seed config: %v", err)
	}

	// ── Build channel keystore ──────────────────────────────────────────────
	entries := make(map[string][]keystore.Entry)

	// Hashtag-derived channels: secret = SHA256("#tag")[:16], hash = SHA256(secret)[0]
	for _, tag := range cfg.ChannelKeys.Hashtags {
		secret, channelHash, fingerprint := keystore.DeriveHashtagKey(tag)
		hashHex := fmt.Sprintf("%02x", channelHash)
		entry := keystore.Entry{
			Key:         secret,
			Fingerprint: fingerprint,
			Hashtag:     tag,
			Name:        "#" + tag,
		}
		if !entryExists(entries[hashHex], entry) {
			entries[hashHex] = append(entries[hashHex], entry)
			log.Printf("config: loaded hashtag channel #%s (hash=%s)", tag, hashHex)
		}
	}

	// Explicit keys: hash provided directly, key is hex-encoded
	for hashHex, keyCfg := range cfg.ChannelKeys.Keys {
		key, err := hex.DecodeString(keyCfg.Key)
		if err != nil {
			log.Printf("warning: invalid channel key for hash %s, skipping: %v", hashHex, err)
			continue
		}
		entry := keystore.Entry{
			Key:         key,
			Fingerprint: keystore.Fingerprint(key),
			Name:        keyCfg.Name,
		}
		if !entryExists(entries[hashHex], entry) {
			entries[hashHex] = append(entries[hashHex], entry)
			log.Printf("config: loaded explicit channel key for hash %s name=%q", hashHex, keyCfg.Name)
		}
	}

	keys := keystore.NewMapKeyStore(entries)

	broker1 := ingest.New(
		ingest.Config{
			BrokerName: "mqtt1",
			URL:        mustEnv("MQTT_BROKER_1_URL"),
			Username:   mustEnv("MQTT_BROKER_1_USERNAME"),
			Password:   mustEnv("MQTT_BROKER_1_PASSWORD"),
		},
		store,
		h,
		keys,
	)

	broker2 := ingest.New(
		ingest.Config{
			BrokerName: "mqtt2",
			URL:        mustEnv("MQTT_BROKER_2_URL"),
			Username:   mustEnv("MQTT_BROKER_2_USERNAME"),
			Password:   mustEnv("MQTT_BROKER_2_PASSWORD"),
		},
		store,
		h,
		keys,
	)

	go broker1.Start(ctx)
	go broker2.Start(ctx)

	// ── HTTP server ──────────────────────────────────────────────────────────
	r := router.New(h, store)

	srv := &http.Server{
		Addr:    addr,
		Handler: r,
	}

	go func() {
		fmt.Printf("Tower listening on %s\n", addr)
		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			log.Fatalf("server error: %v", err)
		}
	}()

	// ── Graceful shutdown ────────────────────────────────────────────────────
	quit := make(chan os.Signal, 1)
	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
	<-quit

	log.Println("shutting down...")
	cancel() // stops ingest workers
	if err := srv.Shutdown(context.Background()); err != nil {
		log.Printf("server shutdown error: %v", err)
	}
}

// entryExists checks if an identical key entry already exists in the slice
// to prevent loading duplicate key/hash pairs from config.
func entryExists(entries []keystore.Entry, e keystore.Entry) bool {
	for _, existing := range entries {
		if bytes.Equal(existing.Key, e.Key) {
			return true
		}
	}
	return false
}

// mustEnv returns the value of an env var and logs a warning if it is unset.
// Callers that require the value to be non-empty should fatal themselves;
// ingest workers tolerate missing broker config and will fail on connect instead.
func mustEnv(key string) string {
	v := os.Getenv(key)
	if v == "" {
		log.Printf("warning: %s is not set", key)
	}
	return v
}