at-container-registry/cmd/hold/main.go

package main

import (
	"context"
	"fmt"
	"log"
	"net/http"

	"atcr.io/pkg/hold"
	"atcr.io/pkg/hold/oci"
	"atcr.io/pkg/hold/pds"
	"atcr.io/pkg/s3"

	// Import storage drivers
	"github.com/distribution/distribution/v3/registry/storage/driver/factory"
	_ "github.com/distribution/distribution/v3/registry/storage/driver/filesystem"
	_ "github.com/distribution/distribution/v3/registry/storage/driver/s3-aws"

	"github.com/go-chi/chi/v5"
)

func main() {
	// Load configuration from environment variables
	cfg, err := hold.LoadConfigFromEnv()
	if err != nil {
		log.Fatalf("Failed to load config: %v", err)
	}

	// Initialize embedded PDS if database path is configured
	// This must happen before creating HoldService since service needs PDS for authorization
	var holdPDS *pds.HoldPDS
	var xrpcHandler *pds.XRPCHandler
	var broadcaster *pds.EventBroadcaster
	if cfg.Database.Path != "" {
		// Generate did:web from public URL
		holdDID := pds.GenerateDIDFromURL(cfg.Server.PublicURL)
		log.Printf("Initializing embedded PDS with DID: %s", holdDID)

		// Initialize PDS with carstore and keys
		ctx := context.Background()
		holdPDS, err = pds.NewHoldPDS(ctx, holdDID, cfg.Server.PublicURL, cfg.Database.Path, cfg.Database.KeyPath)
		if err != nil {
			log.Fatalf("Failed to initialize embedded PDS: %v", err)
		}

		// Bootstrap PDS with captain record and hold owner as first crew member
		if err := holdPDS.Bootstrap(ctx, cfg.Registration.OwnerDID, cfg.Server.Public, cfg.Registration.AllowAllCrew); err != nil {
			log.Fatalf("Failed to bootstrap PDS: %v", err)
		}

		// Create event broadcaster for subscribeRepos firehose
		broadcaster = pds.NewEventBroadcaster(holdDID, 100) // Keep 100 events for backfill

		// Wire up repo event handler to broadcaster
		holdPDS.RepomgrRef().SetEventHandler(broadcaster.SetRepoEventHandler(), true)

		log.Printf("Embedded PDS initialized successfully with firehose enabled")
	} else {
		log.Fatalf("Database path is required for embedded PDS authorization")
	}

	// Create blob store adapter and XRPC handlers
	var ociHandler *oci.XRPCHandler
	if holdPDS != nil {
		// Create storage driver from config
		ctx := context.Background()
		driver, err := factory.Create(ctx, cfg.Storage.Type(), cfg.Storage.Parameters())
		if err != nil {
			log.Fatalf("failed to create storage driver: %v", err)
			return
		}

		s3Service, err := s3.NewS3Service(cfg.Storage.Parameters(), cfg.Server.DisablePresignedURLs, cfg.Storage.Type())
		if err != nil {
			log.Fatalf("Failed to create s3 service: %v", err)
		}

		// Create PDS XRPC handler (ATProto endpoints)
		xrpcHandler = pds.NewXRPCHandler(holdPDS, *s3Service, driver, broadcaster, nil)

		// Create OCI XRPC handler (multipart upload endpoints)
		ociHandler = oci.NewXRPCHandler(holdPDS, *s3Service, driver, cfg.Server.DisablePresignedURLs, nil)
	}

	// Setup HTTP routes with chi router
	r := chi.NewRouter()

	// Root page
	r.Get("/", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "text/plain")
		fmt.Fprintf(w, "This is a hold server. More info at https://atcr.io")
	})

	// Register XRPC/ATProto PDS endpoints if PDS is initialized
	if xrpcHandler != nil {
		log.Printf("Registering ATProto PDS endpoints")
		xrpcHandler.RegisterHandlers(r)
	}

	// Register OCI multipart upload endpoints
	if ociHandler != nil {
		log.Printf("Registering OCI multipart upload endpoints")
		ociHandler.RegisterHandlers(r)
	}

	// Create server
	server := &http.Server{
		Addr:         cfg.Server.Addr,
		Handler:      r,
		ReadTimeout:  cfg.Server.ReadTimeout,
		WriteTimeout: cfg.Server.WriteTimeout,
	}

	// Start server in goroutine so we can do auto-registration after it's running
	serverErr := make(chan error, 1)
	go func() {
		log.Printf("Starting hold service on %s", cfg.Server.Addr)
		if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			serverErr <- err
		}
	}()

	// Wait for server error or shutdown
	if err := <-serverErr; err != nil {
		log.Fatalf("Server failed: %v", err)
	}
}