Port postgres to Go and install the extensions a consumer asks for
letta crash-loops on 'type "vector" does not exist': pgvector is not a trusted extension, so only the provider's superuser can create it, and the provisioner never did. A contribution may now name extensions; the provider creates each (IF NOT EXISTS, available ones only) in the consumer's database on every pass. Go per the standing rule for a TypeScript module that changes. letta asks for vector.
This commit is contained in:
@@ -0,0 +1,89 @@
|
||||
// postgres-provider: postgres's code, one binary the node's runtime launches and speaks MCP to over
|
||||
// stdio through the Go SDK (novox/hq ADR 0188, 0193). It serves postgres's tools and the mesh-store
|
||||
// seat's verbs and, beside them, runs long: the provisioner that makes postgres the provider of the
|
||||
// mesh `postgres-database` interface, and an audit line for each database it provisions or withdraws.
|
||||
//
|
||||
// stdout is the MCP channel; everything this module says, it says on stderr.
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
)
|
||||
|
||||
func say(format string, args ...any) {
|
||||
fmt.Fprintf(os.Stderr, "[postgres] "+format+"\n", args...)
|
||||
}
|
||||
|
||||
func main() {
|
||||
pg, err := ClientFromEnv(os.Getenv)
|
||||
if err != nil {
|
||||
// Without the server there is nothing to serve and nothing to provision; said, not fatal,
|
||||
// so the runtime does not restart a process that cannot do better.
|
||||
say("%v; serving no tools and provisioning nothing", err)
|
||||
if err := stdio.Serve("", nil); err != nil {
|
||||
say("%v", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
return
|
||||
}
|
||||
if receives := os.Getenv("MESH_RECEIVES"); receives == "" {
|
||||
say("MESH_RECEIVES is not set — the provisioner cannot run without it")
|
||||
} else {
|
||||
h := &Harness{
|
||||
Resource: "postgres-database",
|
||||
Receives: receives,
|
||||
Adapter: provisioner{pg: pg, announce: announce},
|
||||
Log: func(format string, args ...any) { fmt.Fprintf(os.Stderr, format+"\n", args...) },
|
||||
}
|
||||
go h.Run(context.Background())
|
||||
}
|
||||
go audit()
|
||||
if err := stdio.Serve("", Tools(pg)); err != nil {
|
||||
say("%v", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
// announce emits a lifecycle event without letting a broker hiccup fail the provisioning itself.
|
||||
func announce(event string, body map[string]string) {
|
||||
if err := stdio.Emit(event, body); err != nil {
|
||||
say("emit %s failed: %v", event, err)
|
||||
}
|
||||
}
|
||||
|
||||
// audit keeps a line for each database granted and withdrawn — observability the provider is best
|
||||
// placed to log. The events are its own, read back from its consumer (`consumes`).
|
||||
func audit() {
|
||||
subscribe := func() error {
|
||||
return stdio.Subscribe("postgres.database.*", func(e stdio.Envelope) error {
|
||||
var body struct {
|
||||
Consumer string `json:"consumer"`
|
||||
Database string `json:"database"`
|
||||
}
|
||||
_ = json.Unmarshal(e.Body, &body)
|
||||
switch e.Key {
|
||||
case "postgres.database.provisioned":
|
||||
say("database provisioned for %s (db %s)", body.Consumer, body.Database)
|
||||
case "postgres.database.deprovisioned":
|
||||
say("database withdrawn, kept (db %s)", body.Database)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
// Asked again until the runtime takes it: the first ask may come before Serve is running.
|
||||
for wait := time.Second; ; wait = min(wait*2, time.Minute) {
|
||||
if err := subscribe(); err == nil {
|
||||
say("auditing database lifecycle events")
|
||||
return
|
||||
} else if wait >= 8*time.Second {
|
||||
say("subscribing to the lifecycle events failed, will retry: %v", err)
|
||||
}
|
||||
time.Sleep(wait)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user