96 lines
3.4 KiB
Go
96 lines
3.4 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"flag"
|
|
"fmt"
|
|
"net/http"
|
|
"os"
|
|
"time"
|
|
|
|
"git.cin.su/m/Assistant/adapters/agentbridge"
|
|
"git.cin.su/m/Assistant/adapters/postgres"
|
|
secretcontracts "git.cin.su/m/Assistant/contracts/secrets"
|
|
"git.cin.su/m/Assistant/contracts/work"
|
|
"git.cin.su/m/Assistant/modules/work/application"
|
|
"git.cin.su/m/Assistant/modules/work/ports"
|
|
platformsecrets "git.cin.su/m/Assistant/platform/secrets"
|
|
)
|
|
|
|
func main() {
|
|
workerID := flag.String("worker-id", "agent-worker-1", "durable worker identity")
|
|
staleAfter := flag.Duration("stale-after", 10*time.Minute, "age after which a claimed item is recovered")
|
|
leaseTTL := flag.Duration("lease-ttl", 15*time.Minute, "dry-run lease duration")
|
|
pollInterval := flag.Duration("poll-interval", 5*time.Second, "database-owned idle wake-up interval")
|
|
once := flag.Bool("once", false, "run one cycle and exit")
|
|
flag.Parse()
|
|
|
|
password, err := resolveDatabasePassword("WORK_POSTGRES_PASSWORD", "WORK_POSTGRES_CREDENTIAL_REF", secretcontracts.RuntimeConsumerWorkWorker)
|
|
if err != nil {
|
|
fmt.Fprintln(os.Stderr, err)
|
|
os.Exit(1)
|
|
}
|
|
repository, err := postgres.OpenWork(context.Background(), postgres.ContinuityConfig{Host: os.Getenv("POSTGRES_HOST"), Port: os.Getenv("POSTGRES_PORT"), Database: os.Getenv("POSTGRES_DATABASE"), User: os.Getenv("WORK_POSTGRES_USER"), Password: password})
|
|
if err != nil {
|
|
fmt.Fprintln(os.Stderr, err)
|
|
os.Exit(1)
|
|
}
|
|
defer repository.Close()
|
|
var executor ports.AgentExecutor
|
|
if bridgeURL := os.Getenv("AGENT_EXECUTION_BRIDGE_URL"); bridgeURL != "" {
|
|
executor, err = agentbridge.NewHTTPExecutor(bridgeURL, &http.Client{Timeout: *leaseTTL})
|
|
if err != nil {
|
|
fmt.Fprintln(os.Stderr, err)
|
|
os.Exit(1)
|
|
}
|
|
} else {
|
|
fmt.Fprintln(os.Stderr, "agent execution bridge is not configured; using legacy dry-run dispatch")
|
|
}
|
|
ctx := context.Background()
|
|
for {
|
|
var result application.WorkerCycleResult
|
|
var cycleErr error
|
|
if executor == nil {
|
|
result, cycleErr = application.ExecuteWorkerCycle(ctx, repository, *workerID, work.AgentFencingToken(1), time.Now().UTC(), *staleAfter, *leaseTTL)
|
|
} else {
|
|
result, cycleErr = application.ExecuteWorkerCycleWithExecutor(ctx, repository, executor, *workerID, work.AgentFencingToken(1), time.Now().UTC(), *staleAfter, *leaseTTL)
|
|
}
|
|
if cycleErr != nil {
|
|
fmt.Fprintln(os.Stderr, cycleErr)
|
|
if *once {
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
if cycleErr == nil && (result.Completed || result.RecoveredClaims > 0) {
|
|
_ = json.NewEncoder(os.Stdout).Encode(result)
|
|
}
|
|
if *once {
|
|
return
|
|
}
|
|
if err := repository.WaitForDispatch(ctx, *pollInterval); err != nil {
|
|
fmt.Fprintln(os.Stderr, err)
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
}
|
|
|
|
func resolveDatabasePassword(name, refName string, consumer secretcontracts.RuntimeConsumer) (string, error) {
|
|
if _, ok := os.LookupEnv(name); ok {
|
|
return "", fmt.Errorf("%s direct environment secret is forbidden", name)
|
|
}
|
|
path, ref := os.Getenv(name+"_FILE"), os.Getenv(refName)
|
|
if path == "" || ref == "" {
|
|
return "", fmt.Errorf("%s_FILE and %s are required", name, refName)
|
|
}
|
|
root := os.Getenv("SECRET_BACKEND_ROOT")
|
|
if root == "" {
|
|
root = "/run/secrets"
|
|
}
|
|
resolver, err := platformsecrets.NewFileResolver(root)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return resolver.ResolveFile(secretcontracts.SecretFileRef{CredentialRef: secretcontracts.CredentialRef(ref), Purpose: secretcontracts.SecretPurposeDatabasePassword, Consumer: consumer, Path: path})
|
|
}
|