Assistant/cmd/continuity-runtime/main.go

package main

import (
	"context"
	"encoding/json"
	"errors"
	"log"
	"net/http"
	"os"
	"strings"
	"time"

	"git.cin.su/m/Assistant/adapters/postgres"
	"git.cin.su/m/Assistant/contracts/continuity"
	"git.cin.su/m/Assistant/platform/ids"
	"git.cin.su/m/Assistant/runtime/health"
)

func main() {
	addr := os.Getenv("LISTEN_ADDR")
	if addr == "" {
		addr = ":8081"
	}
	repository, err := postgres.OpenContinuity(context.Background(), postgres.ContinuityConfig{Host: os.Getenv("POSTGRES_HOST"), Port: os.Getenv("POSTGRES_PORT"), Database: os.Getenv("POSTGRES_DATABASE"), User: os.Getenv("POSTGRES_USER"), Password: os.Getenv("POSTGRES_PASSWORD")})
	if err != nil {
		log.Fatal(err)
	}
	defer repository.Close()
	identityClient, err := newCreatorIdentityClient(envOr("CORE_RUNTIME_URL", "http://core-runtime:8080"), &http.Client{Timeout: 5 * time.Second})
	if err != nil {
		log.Fatal(err)
	}
	mux := http.NewServeMux()
	mux.Handle("/healthz", health.Handler{Runtime: "continuity-runtime", SemanticGeneration: 1, ArchitectureGeneration: 1})
	readinessClient := &http.Client{Timeout: 2 * time.Second}
	mux.Handle("/readyz", health.ReadinessHandler{Runtime: "continuity-runtime", Probes: []health.Probe{
		health.TCPProbe("postgres", envOr("POSTGRES_HOST", "postgres"), envOr("POSTGRES_PORT", "5432")),
		health.HTTPProbe(readinessClient, strings.TrimRight(envOr("CORE_RUNTIME_URL", "http://core-runtime:8080"), "/")+"/healthz"),
		health.HTTPProbe(readinessClient, strings.TrimRight(envOr("COMMIT_RUNTIME_URL", "http://commit-runtime:8082"), "/")+"/healthz"),
	}})
	executorToken := os.Getenv("COMMIT_EXECUTOR_TOKEN")
	if executorToken == "" {
		log.Fatal("COMMIT_EXECUTOR_TOKEN is required")
	}
	commitClient, err := newDueNotificationClient(envOr("COMMIT_RUNTIME_URL", "http://commit-runtime:8082"), executorToken, &http.Client{Timeout: 10 * time.Second})
	if err != nil {
		log.Fatal(err)
	}
	workerContext, cancelWorker := context.WithCancel(context.Background())
	defer cancelWorker()
	dueReminderWorker{repository: repository, commit: commitClient, now: time.Now, newID: ids.New, interval: 5 * time.Second, limit: 10}.Start(workerContext)
	registerInternalExecutorRoutes(mux, repository, executorToken)
	registerCreatorResponseRoute(mux, identityClient, repository, time.Now)
	mux.HandleFunc("POST /v1/prospective-items", func(w http.ResponseWriter, r *http.Request) {
		var item continuity.ProspectiveItem
		if err := json.NewDecoder(r.Body).Decode(&item); err != nil {
			http.Error(w, "invalid prospective item", http.StatusBadRequest)
			return
		}
		if err := repository.Save(r.Context(), item); err != nil {
			http.Error(w, "save prospective item", http.StatusInternalServerError)
			return
		}
		w.WriteHeader(http.StatusCreated)
	})
	registerCreatorReminderRoute(mux, identityClient, repository)
	mux.HandleFunc("GET /v1/prospective-items/{id}", func(w http.ResponseWriter, r *http.Request) {
		item, err := repository.Load(r.Context(), continuity.ProspectiveItemID(r.PathValue("id")))
		if err != nil {
			http.Error(w, "prospective item not found", http.StatusNotFound)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(item)
	})
	mux.HandleFunc("GET /v1/creator/prospective-items/{id}", func(w http.ResponseWriter, r *http.Request) {
		session, err := identityClient.Session(r.Context(), r.Header.Get("Authorization"))
		if errors.Is(err, errCreatorSessionDenied) {
			http.Error(w, "creator session denied", http.StatusUnauthorized)
			return
		}
		if err != nil {
			http.Error(w, "identity service unavailable", http.StatusServiceUnavailable)
			return
		}
		item, err := repository.Load(r.Context(), continuity.ProspectiveItemID(r.PathValue("id")))
		if err != nil {
			http.Error(w, "prospective item not found", http.StatusNotFound)
			return
		}
		if item.OwnerPrincipalID != session.PrincipalID {
			http.Error(w, "prospective item denied", http.StatusForbidden)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(item)
	})
	mux.HandleFunc("GET /v1/creator/prospective-items", func(w http.ResponseWriter, r *http.Request) {
		session, err := identityClient.Session(r.Context(), r.Header.Get("Authorization"))
		if errors.Is(err, errCreatorSessionDenied) {
			http.Error(w, "creator session denied", http.StatusUnauthorized)
			return
		}
		if err != nil {
			http.Error(w, "identity service unavailable", http.StatusServiceUnavailable)
			return
		}
		limit, err := parseListLimit(r.URL.Query().Get("limit"))
		if err != nil {
			http.Error(w, "invalid list limit", http.StatusBadRequest)
			return
		}
		items, err := repository.ByOwner(r.Context(), session.PrincipalID, limit)
		if err != nil {
			http.Error(w, "prospective items unavailable", http.StatusInternalServerError)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(struct {
			Items []continuity.ProspectiveItem
		}{Items: items})
	})
	registerCreatorAttentionRoutes(mux, identityClient, repository)
	log.Printf("continuity-runtime listening on %s", addr)
	if err := http.ListenAndServe(addr, mux); err != nil {
		log.Fatal(err)
	}
}

func envOr(name, fallback string) string {
	if value := os.Getenv(name); value != "" {
		return value
	}
	return fallback
}