Assistant/cmd/commit-runtime/main.go

package main

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

	"git.cin.su/m/Assistant/adapters/continuityexecutor"
	"git.cin.su/m/Assistant/adapters/postgres"
	"git.cin.su/m/Assistant/contracts/authority"
	commitcontract "git.cin.su/m/Assistant/contracts/commit"
	"git.cin.su/m/Assistant/contracts/identity"
	authoritydomain "git.cin.su/m/Assistant/modules/authority/domain"
	commitdomain "git.cin.su/m/Assistant/modules/commit/domain"
	"git.cin.su/m/Assistant/platform/ids"
	"git.cin.su/m/Assistant/runtime/health"
)

type authorizationRequest struct {
	Proposal     commitcontract.ActionProposal
	AuthorizedID commitcontract.AuthorizedActionID
	GrantID      authority.GrantID
	PrincipalID  identity.PrincipalID
}
type authorityValidator struct{}

func (authorityValidator) Validate(context authority.AuthorizationContext) error {
	return authoritydomain.ValidateGrant(context.Binding, context.Scope, context.Grant)
}

func main() {
	addr := os.Getenv("LISTEN_ADDR")
	if addr == "" {
		addr = ":8082"
	}
	config := postgres.ContinuityConfig{Host: os.Getenv("POSTGRES_HOST"), Port: os.Getenv("POSTGRES_PORT"), Database: os.Getenv("POSTGRES_DATABASE")}
	config.User, config.Password = os.Getenv("POSTGRES_USER"), os.Getenv("POSTGRES_PASSWORD")
	repository, err := postgres.OpenCommit(context.Background(), config)
	if err != nil {
		log.Fatal(err)
	}
	defer repository.Close()
	config.User, config.Password = os.Getenv("AUTHORITY_POSTGRES_USER"), os.Getenv("AUTHORITY_POSTGRES_PASSWORD")
	authorityRepository, err := postgres.OpenAuthority(context.Background(), config)
	if err != nil {
		log.Fatal(err)
	}
	defer authorityRepository.Close()
	coreURL := os.Getenv("CORE_RUNTIME_URL")
	if coreURL == "" {
		log.Fatal("CORE_RUNTIME_URL is required")
	}
	executorToken := os.Getenv("COMMIT_EXECUTOR_TOKEN")
	if executorToken == "" {
		log.Fatal("COMMIT_EXECUTOR_TOKEN is required")
	}
	continuityExecutor, err := continuityexecutor.New(envOr("CONTINUITY_RUNTIME_URL", "http://continuity-runtime:8081"), executorToken, &http.Client{Timeout: 5 * time.Second})
	if err != nil {
		log.Fatal(err)
	}
	creatorHTTPClient := &http.Client{Timeout: 5 * time.Second}
	creatorCore, err := newCreatorCoreClient(coreURL, creatorHTTPClient)
	if err != nil {
		log.Fatal(err)
	}
	creatorContinuity, err := newCreatorContinuityClient(envOr("CONTINUITY_RUNTIME_URL", "http://continuity-runtime:8081"), creatorHTTPClient)
	if err != nil {
		log.Fatal(err)
	}
	notificationSecret, err := readCommitSecretFile("NOTIFICATION_WRITE_TOKEN_FILE")
	if err != nil {
		log.Fatal(err)
	}
	notificationSecretRef := os.Getenv("NOTIFICATION_WRITE_SECRET_REF")
	if notificationSecretRef == "" {
		log.Fatal("NOTIFICATION_WRITE_SECRET_REF is required")
	}
	dispatcher := operationDispatcher{continuityExecutor: continuityExecutor, continuityReconciler: continuityExecutor, bindings: creatorCore, secretRef: notificationSecretRef, secret: notificationSecret, client: creatorHTTPClient}
	mux := http.NewServeMux()
	mux.Handle("/healthz", health.Handler{Runtime: "commit-runtime", SemanticGeneration: 1, ArchitectureGeneration: 1})
	registerExecutionRoutes(mux, repository, dispatcher, dispatcher, executorToken)
	registerCreatorCommandRoutes(mux, creatorCommandHandler{
		core: creatorCore, continuity: creatorContinuity, authority: authorityRepository, repository: repository,
		executor: dispatcher, reconciler: dispatcher, newID: ids.New, now: time.Now,
	})
	registerInternalNotificationRoutes(mux, internalNotificationHandler{
		core: creatorCore, continuity: creatorContinuity, authority: authorityRepository, repository: repository,
		executor: dispatcher, newID: ids.New, now: time.Now,
	}, executorToken)
	mux.HandleFunc("POST /v1/authorized-actions", func(w http.ResponseWriter, r *http.Request) {
		var request authorizationRequest
		if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
			http.Error(w, "invalid authorization request", 400)
			return
		}
		grant, scope, err := authorityRepository.LoadGrant(r.Context(), request.GrantID)
		if err != nil {
			http.Error(w, "authority grant not found", http.StatusConflict)
			return
		}
		binding, err := loadJSON[identity.PrincipalBinding](r.Context(), coreURL+"/v1/principal-bindings/"+url.PathEscape(string(request.Proposal.ActorID))+"/"+url.PathEscape(string(request.PrincipalID)))
		if err != nil {
			http.Error(w, "identity binding unavailable", http.StatusConflict)
			return
		}
		target, err := loadJSON[struct {
			TaskID   string
			GoalID   string
			ScopeID  string
			Revision uint64
		}](r.Context(), coreURL+"/v1/tasks/"+url.PathEscape(string(request.Proposal.Target.TaskID)))
		if err != nil || target.TaskID != string(request.Proposal.Target.TaskID) || target.GoalID != string(request.Proposal.Target.GoalID) || target.ScopeID != string(request.Proposal.Target.ScopeID) || target.Revision != uint64(request.Proposal.Target.Revision) {
			http.Error(w, "work target is stale or unavailable", http.StatusConflict)
			return
		}
		action, err := commitdomain.Authorize(request.Proposal, commitdomain.AuthorizationContext{Binding: binding, Scope: scope, Grant: grant}, authorityValidator{}, request.AuthorizedID, time.Now().UnixMilli())
		if err != nil {
			http.Error(w, "authorization denied", http.StatusConflict)
			return
		}
		if err := repository.SaveAuthorized(r.Context(), action); err != nil {
			http.Error(w, "save authorized action", 500)
			return
		}
		w.WriteHeader(http.StatusCreated)
	})
	mux.HandleFunc("POST /v1/action-attempts", func(w http.ResponseWriter, r *http.Request) {
		var attempt commitcontract.ActionAttempt
		if err := json.NewDecoder(r.Body).Decode(&attempt); err != nil {
			http.Error(w, "invalid action attempt", 400)
			return
		}
		if attempt.State != commitcontract.ActionUnknownOutcome && attempt.State != commitcontract.ActionSucceeded && attempt.State != commitcontract.ActionFailed {
			http.Error(w, "attempt must have terminal outcome", http.StatusConflict)
			return
		}
		if err := repository.SaveAttempt(r.Context(), attempt); err != nil {
			http.Error(w, "save action attempt", 500)
			return
		}
		w.WriteHeader(http.StatusCreated)
	})
	log.Printf("commit-runtime listening on %s", addr)
	if err := http.ListenAndServe(addr, mux); err != nil {
		log.Fatal(err)
	}
}

func readCommitSecretFile(name string) (string, error) {
	path := os.Getenv(name)
	if path == "" {
		return "", fmt.Errorf("%s is required", name)
	}
	value, err := os.ReadFile(path)
	if err != nil {
		return "", fmt.Errorf("read %s: %w", name, err)
	}
	secret := strings.TrimSpace(string(value))
	if secret == "" {
		return "", fmt.Errorf("%s is empty", name)
	}
	return secret, nil
}

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

func loadJSON[T any](ctx context.Context, endpoint string) (T, error) {
	var value T
	request, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
	if err != nil {
		return value, err
	}
	response, err := http.DefaultClient.Do(request)
	if err != nil {
		return value, err
	}
	defer response.Body.Close()
	if response.StatusCode != http.StatusOK {
		return value, fmt.Errorf("remote query status %d", response.StatusCode)
	}
	return value, json.NewDecoder(response.Body).Decode(&value)
}