mrhid6/vantage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 22 Wiki Activity

refactor(server): rename Org to Instance

Browse Source 
Adds migration 0004_org_to_instance, the ScopedCollections list, the
AssertNoScopedCollectionMissed boot check, and moves EnsureAuthIndexes into
its own file.

Two ordering constraints the rename exposed, both now enforced and commented:

- 0004 must run BEFORE EnsureAuthIndexes. The index builder creates
  instances.slug, which would create an empty instances collection and make
  0004 refuse to rename orgs onto an existing target.
- Migrations 0001 to 0003 run BEFORE 0004 and still read and write org_id, so
  they use a private legacyOrg struct rather than shared/models.
This commit is contained in:
mrhid6
2026-07-24 13:58:41 +01:00
parent 43a2fdb3a0
commit 4f041d2f4b
61 changed files with 895 additions and 967 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/services/audit.go
+10 -10
View File
@@ -11,25 +11,25 @@ import (
"go.mongodb.org/mongo-driver/v2/mongo/options"
)
func LogEvent(orgID, eventType, actor, serverID, keyID, details string) {
func LogEvent(instanceID, eventType, actor, serverID, keyID, details string) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
event := models.AuditEvent{
OrgID: orgID,
EventType: eventType,
Actor: actor,
ServerID: serverID,
KeyID: keyID,
Details: details,
CreatedAt: time.Now(),
InstanceID: instanceID,
EventType: eventType,
Actor: actor,
ServerID: serverID,
KeyID: keyID,
Details: details,
CreatedAt: time.Now(),
}
if _, err := db.Col("audit_logs").InsertOne(ctx, event); err != nil {
log.Printf("audit log error: %v", err)
}
}
func ListAuditEvents(orgID string, limit int64) ([]models.AuditEvent, error) {
func ListAuditEvents(instanceID string, limit int64) ([]models.AuditEvent, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
@@ -37,7 +37,7 @@ func ListAuditEvents(orgID string, limit int64) ([]models.AuditEvent, error) {
SetSort(bson.D{{Key: "created_at", Value: -1}}).
SetLimit(limit)
cursor, err := db.Col("audit_logs").Find(ctx, bson.M{"org_id": orgID}, opts)
cursor, err := db.Col("audit_logs").Find(ctx, bson.M{"instance_id": instanceID}, opts)
if err != nil {
return nil, err
}
server/internal/services/channels.go
+16 -20
View File
@@ -13,10 +13,10 @@ import (
"go.mongodb.org/mongo-driver/v2/mongo/options"
)
func ListChannels(orgID string) ([]models.NotificationChannel, error) {
func ListChannels(instanceID string) ([]models.NotificationChannel, error) {
ctx, cancel := monCtx()
defer cancel()
cur, err := db.Col("notification_channels").Find(ctx, bson.M{"org_id": orgID}, options.Find().SetSort(bson.M{"created_at": 1}))
cur, err := db.Col("notification_channels").Find(ctx, bson.M{"instance_id": instanceID}, options.Find().SetSort(bson.M{"created_at": 1}))
if err != nil {
return nil, err
}
@@ -27,11 +27,11 @@ func ListChannels(orgID string) ([]models.NotificationChannel, error) {
return out, nil
}
func GetChannel(orgID, channelID string) (*models.NotificationChannel, error) {
func GetChannel(instanceID, channelID string) (*models.NotificationChannel, error) {
ctx, cancel := monCtx()
defer cancel()
var ch models.NotificationChannel
err := db.Col("notification_channels").FindOne(ctx, bson.M{"channel_id": channelID, "org_id": orgID}).Decode(&ch)
err := db.Col("notification_channels").FindOne(ctx, bson.M{"channel_id": channelID, "instance_id": instanceID}).Decode(&ch)
if errors.Is(err, mongo.ErrNoDocuments) {
return nil, nil
}
@@ -41,14 +41,13 @@ func GetChannel(orgID, channelID string) (*models.NotificationChannel, error) {
return &ch, nil
}
func GetChannels(orgID string, channelIDs []string) ([]models.NotificationChannel, error) {
func GetChannels(instanceID string, channelIDs []string) ([]models.NotificationChannel, error) {
if len(channelIDs) == 0 {
return nil, nil
}
ctx, cancel := monCtx()
defer cancel()
cur, err := db.Col("notification_channels").Find(ctx, bson.M{"org_id": orgID, "channel_id": bson.M{"$in": channelIDs}})
cur, err := db.Col("notification_channels").Find(ctx, bson.M{"instance_id": instanceID, "channel_id": bson.M{"$in": channelIDs}})
if err != nil {
return nil, err
}
@@ -59,11 +58,9 @@ func GetChannels(orgID string, channelIDs []string) ([]models.NotificationChanne
return out, nil
}
func validateChannelIDs(orgID string, channelIDs []string) error {
func validateChannelIDs(instanceID string, channelIDs []string) error {
for _, id := range channelIDs {
ch, err := GetChannel(orgID, id)
ch, err := GetChannel(instanceID, id)
if err != nil {
return err
}
@@ -74,10 +71,10 @@ func validateChannelIDs(orgID string, channelIDs []string) error {
return nil
}
func CreateChannel(orgID string, ch *models.NotificationChannel) (*models.NotificationChannel, error) {
func CreateChannel(instanceID string, ch *models.NotificationChannel) (*models.NotificationChannel, error) {
ctx, cancel := monCtx()
defer cancel()
ch.OrgID = orgID
ch.InstanceID = instanceID
ch.ChannelID = uuid.NewString()
ch.CreatedAt = time.Now()
if ch.Config == nil {
@@ -89,23 +86,22 @@ func CreateChannel(orgID string, ch *models.NotificationChannel) (*models.Notifi
return ch, nil
}
func UpdateChannel(orgID, channelID string, upd bson.M) error {
func UpdateChannel(instanceID, channelID string, upd bson.M) error {
ctx, cancel := monCtx()
defer cancel()
_, err := db.Col("notification_channels").UpdateOne(ctx, bson.M{"channel_id": channelID, "org_id": orgID}, bson.M{"$set": upd})
_, err := db.Col("notification_channels").UpdateOne(ctx, bson.M{"channel_id": channelID, "instance_id": instanceID}, bson.M{"$set": upd})
return err
}
func DeleteChannel(orgID, channelID string) error {
func DeleteChannel(instanceID, channelID string) error {
ctx, cancel := monCtx()
defer cancel()
_, err := db.Col("notification_channels").DeleteOne(ctx, bson.M{"channel_id": channelID, "org_id": orgID})
_, err := db.Col("notification_channels").DeleteOne(ctx, bson.M{"channel_id": channelID, "instance_id": instanceID})
return err
}
func TestChannel(orgID, channelID string) error {
ch, err := GetChannel(orgID, channelID)
func TestChannel(instanceID, channelID string) error {
ch, err := GetChannel(instanceID, channelID)
if err != nil {
return err
}
server/internal/services/console.go
+23 -38
View File
@@ -17,7 +17,7 @@ import (
)
func sessionHMACKey() ([]byte, error) {
k, err := encryptionKey()
if err != nil {
return nil, err
@@ -29,7 +29,6 @@ func sessionHMACKey() ([]byte, error) {
func b64(b []byte) string { return base64.RawURLEncoding.EncodeToString(b) }
func SignSessionToken(sessionID string, ttl time.Duration) (string, error) {
key, err := sessionHMACKey()
if err != nil {
@@ -42,7 +41,6 @@ func SignSessionToken(sessionID string, ttl time.Duration) (string, error) {
return payload + "." + b64(mac.Sum(nil)), nil
}
func VerifySessionToken(token string) (string, error) {
parts := strings.Split(token, ".")
if len(parts) != 3 {
@@ -86,10 +84,6 @@ func portOr(v, def int) string {
return strconv.Itoa(v)
}
func BuildGuacParams(srv *models.Server, protocol, sshUser, privateKey, passphrase, rdpUser, rdpPass string) (*GuacParams, error) {
host := srv.IPAddress
switch protocol {
@@ -129,19 +123,19 @@ func BuildGuacParams(srv *models.Server, protocol, sshUser, privateKey, passphra
}
}
func CreateConsoleSession(orgID, serverID, protocol, keyID, user, clientIP string) (*models.ConsoleSession, error) {
func CreateConsoleSession(instanceID, serverID, protocol, keyID, user, clientIP string) (*models.ConsoleSession, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
s := &models.ConsoleSession{
OrgID: orgID,
SessionID: uuid.NewString(),
ServerID: serverID,
Protocol: protocol,
KeyID: keyID,
User: user,
ClientIP: clientIP,
StartedAt: time.Now(),
InstanceID: instanceID,
SessionID: uuid.NewString(),
ServerID: serverID,
Protocol: protocol,
KeyID: keyID,
User: user,
ClientIP: clientIP,
StartedAt: time.Now(),
}
if _, err := db.Col("console_sessions").InsertOne(ctx, s); err != nil {
return nil, err
@@ -149,19 +143,17 @@ func CreateConsoleSession(orgID, serverID, protocol, keyID, user, clientIP strin
return s, nil
}
func GetConsoleSession(orgID, sessionID string) (*models.ConsoleSession, error) {
func GetConsoleSession(instanceID, sessionID string) (*models.ConsoleSession, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
var s models.ConsoleSession
if err := db.Col("console_sessions").FindOne(ctx, bson.M{"session_id": sessionID, "org_id": orgID}).Decode(&s); err != nil {
if err := db.Col("console_sessions").FindOne(ctx, bson.M{"session_id": sessionID, "instance_id": instanceID}).Decode(&s); err != nil {
return nil, err
}
return &s, nil
}
func StashConsoleRDPCreds(orgID, sessionID, username, password string) error {
func StashConsoleRDPCreds(instanceID, sessionID, username, password string) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
u, err := encryptString(username)
@@ -173,17 +165,14 @@ func StashConsoleRDPCreds(orgID, sessionID, username, password string) error {
return err
}
_, err = db.Col("console_sessions").UpdateOne(ctx,
bson.M{"session_id": sessionID, "org_id": orgID},
bson.M{"session_id": sessionID, "instance_id": instanceID},
bson.M{"$set": bson.M{"rdp_user_enc": u, "rdp_pass_enc": p}},
)
return err
}
func ConsumeConsoleRDPCreds(orgID, sessionID string) (username, password string, err error) {
s, err := GetConsoleSession(orgID, sessionID)
func ConsumeConsoleRDPCreds(instanceID, sessionID string) (username, password string, err error) {
s, err := GetConsoleSession(instanceID, sessionID)
if err != nil {
return "", "", err
}
@@ -203,31 +192,27 @@ func ConsumeConsoleRDPCreds(orgID, sessionID string) (username, password string,
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_, _ = db.Col("console_sessions").UpdateOne(ctx,
bson.M{"session_id": sessionID, "org_id": orgID},
bson.M{"session_id": sessionID, "instance_id": instanceID},
bson.M{"$unset": bson.M{"rdp_user_enc": "", "rdp_pass_enc": ""}},
)
return username, password, nil
}
func SetConsoleSSHUser(orgID, sessionID, username string) error {
func SetConsoleSSHUser(instanceID, sessionID, username string) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_, err := db.Col("console_sessions").UpdateOne(ctx,
bson.M{"session_id": sessionID, "org_id": orgID},
bson.M{"session_id": sessionID, "instance_id": instanceID},
bson.M{"$set": bson.M{"ssh_username": username}})
return err
}
func ConsumeSessionToken(orgID, sessionID string) error {
func ConsumeSessionToken(instanceID, sessionID string) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
now := time.Now()
res, err := db.Col("console_sessions").UpdateOne(ctx,
bson.M{"session_id": sessionID, "org_id": orgID, "token_consumed_at": nil},
bson.M{"session_id": sessionID, "instance_id": instanceID, "token_consumed_at": nil},
bson.M{"$set": bson.M{"token_consumed_at": now}},
)
if err != nil {
@@ -239,12 +224,12 @@ func ConsumeSessionToken(orgID, sessionID string) error {
return nil
}
func EndConsoleSession(orgID, sessionID string) error {
func EndConsoleSession(instanceID, sessionID string) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
now := time.Now()
_, err := db.Col("console_sessions").UpdateOne(ctx,
bson.M{"session_id": sessionID, "org_id": orgID, "ended_at": nil},
bson.M{"session_id": sessionID, "instance_id": instanceID, "ended_at": nil},
bson.M{"$set": bson.M{"ended_at": now}},
)
return err
server/internal/services/coreindexes.go
+41
View File
@@ -0,0 +1,41 @@
package services
import (
"context"
"time"
"github.com/mrhid6/vantage/server/internal/db"
"github.com/mrhid6/vantage/shared/indexes"
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo"
"go.mongodb.org/mongo-driver/v2/mongo/options"
)
// EnsureAuthIndexes declares the indexes tenant isolation depends on.
//
// It lives here rather than in migrate.go so that the org-to-instance rename
// could touch it without touching migrations 0001 to 0003, which deliberately
// still speak the pre-rename shape.
//
// It MUST run after MigrateOrgToInstance. Creating the instances.slug index
// first would create an empty instances collection, and migration 0004 refuses
// to rename orgs when instances already exists.
func EnsureAuthIndexes() error {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
// users.email and instances.slug are declared in the shared module so the
// control plane and sitesvc cannot disagree about them.
if err := indexes.EnsureCoreIndexes(ctx, db.Database); err != nil {
return err
}
// instance_oidc is control-plane only, so its index stays here.
if _, err := db.Col("instance_oidc").Indexes().CreateOne(ctx, mongo.IndexModel{
Keys: bson.D{{Key: "instance_id", Value: 1}},
Options: options.Index().SetUnique(true),
}); err != nil {
return err
}
return nil
}
server/internal/services/crypto.go
-3
View File
@@ -22,8 +22,6 @@ func encryptionKey() ([]byte, error) {
return key, nil
}
func encryptString(plaintext string) (string, error) {
key, err := encryptionKey()
if err != nil {
@@ -45,7 +43,6 @@ func encryptString(plaintext string) (string, error) {
return hex.EncodeToString(sealed), nil
}
func decryptString(ciphertextHex string) (string, error) {
key, err := encryptionKey()
if err != nil {
server/internal/services/defaults.go
+7 -13
View File
@@ -13,7 +13,6 @@ import (
"go.mongodb.org/mongo-driver/v2/mongo/options"
)
func DefaultStepsDir() string {
dir := os.Getenv("VANTAGE_DEFAULT_STEPS_DIR")
if dir == "" {
@@ -23,9 +22,6 @@ func DefaultStepsDir() string {
return dir
}
func readDefaultStepFiles() ([]models.WorkflowStep, error) {
matches, err := filepath.Glob(filepath.Join(DefaultStepsDir(), "*.json"))
if err != nil {
@@ -51,9 +47,7 @@ func readDefaultStepFiles() ([]models.WorkflowStep, error) {
return out, nil
}
func SeedDefaultSteps(orgID string) (created, updated int, err error) {
func SeedDefaultSteps(instanceID string) (created, updated int, err error) {
steps, err := readDefaultStepFiles()
if err != nil {
return 0, 0, err
@@ -62,7 +56,7 @@ func SeedDefaultSteps(orgID string) (created, updated int, err error) {
defer cancel()
col := db.Col("workflow_steps")
for _, s := range steps {
filter := bson.M{"org_id": orgID, "slug": s.Slug, "source": "default"}
filter := bson.M{"instance_id": instanceID, "slug": s.Slug, "source": "default"}
set := bson.M{
"name": s.Name,
"description": s.Description,
@@ -76,11 +70,11 @@ func SeedDefaultSteps(orgID string) (created, updated int, err error) {
res, uerr := col.UpdateOne(ctx, filter, bson.M{
"$set": set,
"$setOnInsert": bson.M{
"org_id": orgID,
"step_id": uuid.New().String(),
"slug": s.Slug,
"source": "default",
"created_at": time.Now(),
"instance_id": instanceID,
"step_id": uuid.New().String(),
"slug": s.Slug,
"source": "default",
"created_at": time.Now(),
},
}, options.UpdateOne().SetUpsert(true))
if uerr != nil {
server/internal/services/dispatch.go
+2 -22
View File
@@ -17,13 +17,10 @@ type commandDispatcher struct {
channels map[string]chan *pb.ServerCommand
}
var Dispatcher = &commandDispatcher{
channels: make(map[string]chan *pb.ServerCommand),
}
func (d *commandDispatcher) Connect(serverID string) chan *pb.ServerCommand {
ch := make(chan *pb.ServerCommand, 16)
d.mu.Lock()
@@ -32,14 +29,12 @@ func (d *commandDispatcher) Connect(serverID string) chan *pb.ServerCommand {
return ch
}
func (d *commandDispatcher) Disconnect(serverID string) {
d.mu.Lock()
delete(d.channels, serverID)
d.mu.Unlock()
}
func (d *commandDispatcher) IsConnected(serverID string) bool {
d.mu.RLock()
_, ok := d.channels[serverID]
@@ -62,15 +57,10 @@ func (d *commandDispatcher) dispatch(serverID string, cmd *pb.ServerCommand) err
}
}
func DispatchRunStep(serverID, commandID string, cmd *pb.RunStepCmd) error {
return Dispatcher.dispatch(serverID, &pb.ServerCommand{CommandId: commandID, RunStep: cmd})
}
func DispatchCleanupWorkspace(serverID, workspaceID string) {
if !Dispatcher.IsConnected(serverID) {
return
@@ -81,7 +71,6 @@ func DispatchCleanupWorkspace(serverID, workspaceID string) {
})
}
type KeyGenParams struct {
Label string
KeyType string
@@ -90,15 +79,13 @@ type KeyGenParams struct {
Comment string
}
func GetLatestAgentVersion() (string, error) {
giteaHost := os.Getenv("GITEA_HOST")
if giteaHost == "" {
giteaHost = "gitea.example.com"
}
url := fmt.Sprintf("https://%s/api/v1/repos/mrhid6/vantage/releases?limit=20", giteaHost)
resp, err := http.Get(url)
resp, err := http.Get(url)
if err != nil {
return "", fmt.Errorf("fetch releases: %w", err)
}
@@ -122,8 +109,6 @@ func GetLatestAgentVersion() (string, error) {
return "", fmt.Errorf("no agent release found")
}
func DispatchUpdateAgent(serverID string) (string, error) {
if !Dispatcher.IsConnected(serverID) {
return "", fmt.Errorf("agent is not connected to the command stream")
@@ -153,7 +138,6 @@ func DispatchUpdateAgent(serverID string) (string, error) {
return version, nil
}
func DispatchApplyUpdates(serverID string) error {
if !Dispatcher.IsConnected(serverID) {
return fmt.Errorf("agent is not connected to the command stream")
@@ -165,8 +149,6 @@ func DispatchApplyUpdates(serverID string) error {
return Dispatcher.dispatch(serverID, cmd)
}
func DispatchDeleteKey(serverID, label string) {
if !Dispatcher.IsConnected(serverID) {
return
@@ -176,13 +158,11 @@ func DispatchDeleteKey(serverID, label string) {
DeleteKey: &pb.DeleteKeyCmd{Label: label},
}
if err := Dispatcher.dispatch(serverID, cmd); err != nil {
_ = err
}
}
func DispatchGenerateKey(serverID string, p KeyGenParams) (string, error) {
if !Dispatcher.IsConnected(serverID) {
return "", fmt.Errorf("agent is not connected to the command stream")
server/internal/services/org_oidc.go → server/internal/services/instance_oidc.go
+8 -10
View File
@@ -10,32 +10,30 @@ import (
"go.mongodb.org/mongo-driver/v2/mongo/options"
)
func GetOrgOIDC(orgID string) (*models.OrgOIDC, error) {
func GetInstanceOIDC(instanceID string) (*models.OrgOIDC, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
var o models.OrgOIDC
err := db.Col("org_oidc").FindOne(ctx, bson.M{"org_id": orgID}).Decode(&o)
err := db.Col("instance_oidc").FindOne(ctx, bson.M{"instance_id": instanceID}).Decode(&o)
if err != nil {
return nil, err
}
return &o, nil
}
func GetOrgOIDCSecret(orgID string) (string, error) {
o, err := GetOrgOIDC(orgID)
func GetInstanceOIDCSecret(instanceID string) (string, error) {
o, err := GetInstanceOIDC(instanceID)
if err != nil {
return "", err
}
return decryptString(o.ClientSecretEnc)
}
func SaveOrgOIDC(orgID, issuer, clientID, clientSecret string, enabled bool) error {
func SaveOrgOIDC(instanceID, issuer, clientID, clientSecret string, enabled bool) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
set := bson.M{
"org_id": orgID, "issuer": issuer, "client_id": clientID,
"instance_id": instanceID, "issuer": issuer, "client_id": clientID,
"enabled": enabled, "updated_at": time.Now(),
}
if clientSecret != "" {
@@ -45,8 +43,8 @@ func SaveOrgOIDC(orgID, issuer, clientID, clientSecret string, enabled bool) err
}
set["client_secret_enc"] = enc
}
_, err := db.Col("org_oidc").UpdateOne(ctx,
bson.M{"org_id": orgID}, bson.M{"$set": set},
_, err := db.Col("instance_oidc").UpdateOne(ctx,
bson.M{"instance_id": instanceID}, bson.M{"$set": set},
options.UpdateOne().SetUpsert(true))
return err
}
server/internal/services/orgs.go → server/internal/services/instances.go
+25 -30
View File
@@ -13,69 +13,64 @@ import (
"go.mongodb.org/mongo-driver/v2/mongo"
)
func GetOrg(orgID string) (*models.Org, error) {
func GetInstance(instanceID string) (*models.Instance, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
var o models.Org
err := db.Col("orgs").FindOne(ctx, bson.M{"org_id": orgID}).Decode(&o)
var o models.Instance
err := db.Col("instances").FindOne(ctx, bson.M{"instance_id": instanceID}).Decode(&o)
if err != nil {
return nil, err
}
return &o, nil
}
func GetOrgBySlug(slug string) (*models.Org, error) {
func GetInstanceBySlug(slug string) (*models.Instance, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
var o models.Org
err := db.Col("orgs").FindOne(ctx, bson.M{"slug": slug}).Decode(&o)
var o models.Instance
err := db.Col("instances").FindOne(ctx, bson.M{"slug": slug}).Decode(&o)
if err != nil {
return nil, err
}
return &o, nil
}
func ListOrgIDs() ([]string, error) {
func ListInstanceIDs() ([]string, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
cursor, err := db.Col("orgs").Find(ctx, bson.M{})
cursor, err := db.Col("instances").Find(ctx, bson.M{})
if err != nil {
return nil, err
}
defer cursor.Close(ctx)
var orgs []models.Org
var orgs []models.Instance
if err := cursor.All(ctx, &orgs); err != nil {
return nil, err
}
ids := make([]string, 0, len(orgs))
for _, o := range orgs {
ids = append(ids, o.OrgID)
ids = append(ids, o.InstanceID)
}
return ids, nil
}
func CountOrgs() (int64, error) {
func CountInstances() (int64, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
return db.Col("orgs").CountDocuments(ctx, bson.M{})
return db.Col("instances").CountDocuments(ctx, bson.M{})
}
func FirstOrg() (*models.Org, error) {
func FirstInstance() (*models.Instance, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
var o models.Org
if err := db.Col("orgs").FindOne(ctx, bson.M{}).Decode(&o); err != nil {
var o models.Instance
if err := db.Col("instances").FindOne(ctx, bson.M{}).Decode(&o); err != nil {
return nil, err
}
return &o, nil
}
func AdoptOrg(orgID, name string) (*models.Org, error) {
func AdoptInstance(instanceID, name string) (*models.Instance, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
@@ -86,7 +81,7 @@ func AdoptOrg(orgID, name string) (*models.Org, error) {
slug = slug[:provision.MaxSlugLength]
}
if len(slug) >= provision.MinSlugLength && !provision.ReservedSlugs[slug] {
n, err := db.Col("orgs").CountDocuments(ctx, bson.M{"slug": slug, "org_id": bson.M{"$ne": orgID}})
n, err := db.Col("instances").CountDocuments(ctx, bson.M{"slug": slug, "instance_id": bson.M{"$ne": instanceID}})
if err != nil {
return nil, err
}
@@ -95,33 +90,33 @@ func AdoptOrg(orgID, name string) (*models.Org, error) {
}
}
if _, err := db.Col("orgs").UpdateOne(ctx, bson.M{"org_id": orgID}, bson.M{"$set": set}); err != nil {
if _, err := db.Col("instances").UpdateOne(ctx, bson.M{"instance_id": instanceID}, bson.M{"$set": set}); err != nil {
if mongo.IsDuplicateKeyError(err) {
return nil, fmt.Errorf("organization slug already taken")
}
return nil, err
}
return GetOrg(orgID)
return GetInstance(instanceID)
}
// CreateOrg creates an organisation and seeds its default workflow steps.
// CreateInstance creates an organisation and seeds its default workflow steps.
//
// The creation rules live in shared/provision because sitesvc creates
// organisations too. Seeding stays here: shared must not know about workflow
// steps.
func CreateOrg(name string) (*models.Org, error) {
func CreateInstance(name string) (*models.Instance, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
o, err := provision.CreateOrg(ctx, db.Database, name)
o, err := provision.CreateInstance(ctx, db.Database, name)
if err != nil {
return nil, err
}
if created, updated, err := SeedDefaultSteps(o.OrgID); err != nil {
log.Printf("warning: failed to seed default steps for new org %s: %v", o.OrgID, err)
if created, updated, err := SeedDefaultSteps(o.InstanceID); err != nil {
log.Printf("warning: failed to seed default steps for new org %s: %v", o.InstanceID, err)
} else {
log.Printf("default steps seeded for new org %s: %d created, %d updated", o.OrgID, created, updated)
log.Printf("default steps seeded for new org %s: %d created, %d updated", o.InstanceID, created, updated)
}
return o, nil
}
server/internal/services/inventory.go
-2
View File
@@ -9,8 +9,6 @@ import (
"go.mongodb.org/mongo-driver/v2/bson"
)
func StoreInventory(serverID string, r *pb.InventoryReport) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
server/internal/services/keys.go
+33 -34
View File
@@ -36,9 +36,9 @@ func setKeyMeta(k *models.Key) {
k.HasPassphrase = k.PassphraseEncrypted != ""
}
func CreateKey(orgID, label, publicKey, source, generatedByServerID, privateKey, passphrase string) (*models.Key, error) {
func CreateKey(instanceID, label, publicKey, source, generatedByServerID, privateKey, passphrase string) (*models.Key, error) {
key := &models.Key{
OrgID: orgID,
InstanceID: instanceID,
KeyID: uuid.NewString(),
Label: label,
PublicKey: publicKey,
@@ -72,12 +72,12 @@ func CreateKey(orgID, label, publicKey, source, generatedByServerID, privateKey,
return key, nil
}
func GetKey(orgID, keyID string) (*models.Key, error) {
func GetKey(instanceID, keyID string) (*models.Key, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
var key models.Key
err := db.Col("keys").FindOne(ctx, bson.M{"key_id": keyID, "org_id": orgID}).Decode(&key)
err := db.Col("keys").FindOne(ctx, bson.M{"key_id": keyID, "instance_id": instanceID}).Decode(&key)
if err != nil {
return nil, err
}
@@ -85,12 +85,12 @@ func GetKey(orgID, keyID string) (*models.Key, error) {
return &key, nil
}
func GetPrivateKey(orgID, keyID string) (string, error) {
func GetPrivateKey(instanceID, keyID string) (string, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
var key models.Key
if err := db.Col("keys").FindOne(ctx, bson.M{"key_id": keyID, "org_id": orgID}).Decode(&key); err != nil {
if err := db.Col("keys").FindOne(ctx, bson.M{"key_id": keyID, "instance_id": instanceID}).Decode(&key); err != nil {
return "", err
}
if key.PrivateKeyEncrypted == "" {
@@ -118,11 +118,11 @@ type KeyWithCount struct {
AssignedCount int `bson:"-" json:"assigned_count"`
}
func ListKeys(orgID string) ([]KeyWithCount, error) {
func ListKeys(instanceID string) ([]KeyWithCount, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
cursor, err := db.Col("keys").Find(ctx, bson.M{"org_id": orgID})
cursor, err := db.Col("keys").Find(ctx, bson.M{"instance_id": instanceID})
if err != nil {
return nil, err
}
@@ -137,28 +137,28 @@ func ListKeys(orgID string) ([]KeyWithCount, error) {
for _, k := range keys {
setKeyMeta(&k)
count, _ := db.Col("assignments").CountDocuments(ctx, bson.M{
"org_id": orgID,
"key_id": k.KeyID,
"revoked_at": nil,
"instance_id": instanceID,
"key_id": k.KeyID,
"revoked_at": nil,
})
result = append(result, KeyWithCount{Key: k, AssignedCount: int(count)})
}
return result, nil
}
func DeleteKey(orgID, keyID string) error {
func DeleteKey(instanceID, keyID string) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
var key models.Key
if err := db.Col("keys").FindOne(ctx, bson.M{"key_id": keyID, "org_id": orgID}).Decode(&key); err != nil {
if err := db.Col("keys").FindOne(ctx, bson.M{"key_id": keyID, "instance_id": instanceID}).Decode(&key); err != nil {
return err
}
if _, err := db.Col("keys").DeleteOne(ctx, bson.M{"key_id": keyID, "org_id": orgID}); err != nil {
if _, err := db.Col("keys").DeleteOne(ctx, bson.M{"key_id": keyID, "instance_id": instanceID}); err != nil {
return err
}
if _, err := db.Col("assignments").DeleteMany(ctx, bson.M{"key_id": keyID, "org_id": orgID}); err != nil {
if _, err := db.Col("assignments").DeleteMany(ctx, bson.M{"key_id": keyID, "instance_id": instanceID}); err != nil {
return err
}
@@ -168,31 +168,30 @@ func DeleteKey(orgID, keyID string) error {
return nil
}
func AssignKey(orgID, keyID, serverID string) (*models.Assignment, error) {
func AssignKey(instanceID, keyID, serverID string) (*models.Assignment, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if _, err := GetKey(orgID, keyID); err != nil {
if _, err := GetKey(instanceID, keyID); err != nil {
return nil, fmt.Errorf("key not found")
}
if _, err := GetServer(orgID, serverID); err != nil {
if _, err := GetServer(instanceID, serverID); err != nil {
return nil, fmt.Errorf("server not found")
}
var existing models.Assignment
err := db.Col("assignments").FindOne(ctx, bson.M{
"org_id": orgID,
"key_id": keyID,
"server_id": serverID,
"revoked_at": nil,
"instance_id": instanceID,
"key_id": keyID,
"server_id": serverID,
"revoked_at": nil,
}).Decode(&existing)
if err == nil {
return &existing, nil
}
a := &models.Assignment{
OrgID: orgID,
InstanceID: instanceID,
KeyID: keyID,
ServerID: serverID,
AssignedAt: time.Now(),
@@ -204,23 +203,23 @@ func AssignKey(orgID, keyID, serverID string) (*models.Assignment, error) {
return a, nil
}
func RevokeAssignment(orgID, keyID, serverID string) error {
func RevokeAssignment(instanceID, keyID, serverID string) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
now := time.Now()
_, err := db.Col("assignments").UpdateOne(ctx,
bson.M{"org_id": orgID, "key_id": keyID, "server_id": serverID, "revoked_at": nil},
bson.M{"instance_id": instanceID, "key_id": keyID, "server_id": serverID, "revoked_at": nil},
bson.M{"$set": bson.M{"revoked_at": now}},
)
return err
}
func GetAssignmentsForKey(orgID, keyID string) ([]models.Assignment, error) {
func GetAssignmentsForKey(instanceID, keyID string) ([]models.Assignment, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
cursor, err := db.Col("assignments").Find(ctx, bson.M{"org_id": orgID, "key_id": keyID, "revoked_at": nil})
cursor, err := db.Col("assignments").Find(ctx, bson.M{"instance_id": instanceID, "key_id": keyID, "revoked_at": nil})
if err != nil {
return nil, err
}
@@ -238,11 +237,11 @@ type AssignmentWithServer struct {
Server *models.Server `json:"server,omitempty"`
}
func GetAssignmentsWithServers(orgID, keyID string) ([]AssignmentWithServer, error) {
func GetAssignmentsWithServers(instanceID, keyID string) ([]AssignmentWithServer, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
cursor, err := db.Col("assignments").Find(ctx, bson.M{"org_id": orgID, "key_id": keyID})
cursor, err := db.Col("assignments").Find(ctx, bson.M{"instance_id": instanceID, "key_id": keyID})
if err != nil {
return nil, err
}
@@ -257,7 +256,7 @@ func GetAssignmentsWithServers(orgID, keyID string) ([]AssignmentWithServer, err
for _, a := range assignments {
item := AssignmentWithServer{Assignment: a}
var srv models.Server
if err := db.Col("servers").FindOne(ctx, bson.M{"server_id": a.ServerID, "org_id": orgID}).Decode(&srv); err == nil {
if err := db.Col("servers").FindOne(ctx, bson.M{"server_id": a.ServerID, "instance_id": instanceID}).Decode(&srv); err == nil {
item.Server = &srv
}
result = append(result, item)
@@ -270,11 +269,11 @@ type AssignmentWithKey struct {
Key *models.Key `json:"key,omitempty"`
}
func GetAssignmentsWithKeysForServer(orgID, serverID string) ([]AssignmentWithKey, error) {
func GetAssignmentsWithKeysForServer(instanceID, serverID string) ([]AssignmentWithKey, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
cursor, err := db.Col("assignments").Find(ctx, bson.M{"org_id": orgID, "server_id": serverID})
cursor, err := db.Col("assignments").Find(ctx, bson.M{"instance_id": instanceID, "server_id": serverID})
if err != nil {
return nil, err
}
@@ -288,7 +287,7 @@ func GetAssignmentsWithKeysForServer(orgID, serverID string) ([]AssignmentWithKe
result := make([]AssignmentWithKey, 0, len(assignments))
for _, a := range assignments {
var key models.Key
if err := db.Col("keys").FindOne(ctx, bson.M{"key_id": a.KeyID, "org_id": orgID}).Decode(&key); err != nil {
if err := db.Col("keys").FindOne(ctx, bson.M{"key_id": a.KeyID, "instance_id": instanceID}).Decode(&key); err != nil {
continue
}
setKeyMeta(&key)
server/internal/services/migrate.go
+20 -33
View File
@@ -8,47 +8,36 @@ import (
"github.com/google/uuid"
"github.com/mrhid6/vantage/server/internal/db"
"github.com/mrhid6/vantage/server/internal/models"
"github.com/mrhid6/vantage/shared/indexes"
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo"
"go.mongodb.org/mongo-driver/v2/mongo/options"
)
var scopedCollections = []string{
// legacyOrg is the pre-0004 shape of the orgs collection.
//
// Migrations 0001 to 0003 run BEFORE the org-to-instance rename and must keep
// reading and writing org_id in the orgs collection. They deliberately do not
// use shared/models, which has moved on to Instance and instance_id.
type legacyOrg struct {
OrgID string `bson:"org_id"`
Name string `bson:"name"`
Slug string `bson:"slug"`
CreatedAt time.Time `bson:"created_at"`
}
var backfillCollections = []string{
"servers", "keys", "assignments", "secrets",
"workflows", "workflow_steps", "workflow_runs",
"audit_logs", "monitors", "notification_channels",
"console_sessions", "incidents", "monitor_rollups",
}
func EnsureAuthIndexes() error {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
// users.email and orgs.slug are declared in the shared module so the
// control plane and sitesvc cannot disagree about them.
if err := indexes.EnsureCoreIndexes(ctx, db.Database); err != nil {
return err
}
// org_oidc is control-plane only, so its index stays here.
if _, err := db.Col("org_oidc").Indexes().CreateOne(ctx, mongo.IndexModel{
Keys: bson.D{{Key: "org_id", Value: 1}},
Options: options.Index().SetUnique(true),
}); err != nil {
return err
}
return nil
}
func defaultBackfillOrg(ctx context.Context) (*models.Org, error) {
var org models.Org
func defaultBackfillOrg(ctx context.Context) (*legacyOrg, error) {
var org legacyOrg
err := db.Col("orgs").FindOne(ctx, bson.M{"slug": "default"}).Decode(&org)
switch {
case err == nil:
case errors.Is(err, mongo.ErrNoDocuments):
org = models.Org{OrgID: uuid.NewString(), Name: "Default", Slug: "default", CreatedAt: time.Now()}
org = legacyOrg{OrgID: uuid.NewString(), Name: "Default", Slug: "default", CreatedAt: time.Now()}
if _, err := db.Col("orgs").InsertOne(ctx, org); err != nil {
return nil, err
}
@@ -67,9 +56,8 @@ func RunMigrations() error {
return nil
}
needs := false
for _, col := range scopedCollections {
for _, col := range backfillCollections {
n, _ := db.Col(col).CountDocuments(ctx, bson.M{"org_id": bson.M{"$exists": false}})
if n > 0 {
needs = true
@@ -82,7 +70,7 @@ func RunMigrations() error {
if err != nil {
return err
}
for _, col := range scopedCollections {
for _, col := range backfillCollections {
if _, err := db.Col(col).UpdateMany(ctx,
bson.M{"org_id": bson.M{"$exists": false}},
bson.M{"$set": bson.M{"org_id": org.OrgID}},
@@ -105,7 +93,6 @@ func MigrateMissedOrgScopes() error {
return nil
}
missed := []string{"audit_logs", "notification_channels"}
needs := false
for _, col := range missed {
@@ -186,7 +173,7 @@ func MigrateSettingsOrg() error {
n, _ := db.Col("settings").CountDocuments(ctx, bson.M{"org_id": bson.M{"$exists": false}})
if n > 0 {
var org models.Org
var org legacyOrg
orgCount, err := db.Col("orgs").CountDocuments(ctx, bson.M{})
if err != nil {
return err
@@ -197,7 +184,7 @@ func MigrateSettingsOrg() error {
return err
}
case 0:
org = models.Org{OrgID: uuid.NewString(), Name: "Default", Slug: "default", CreatedAt: time.Now()}
org = legacyOrg{OrgID: uuid.NewString(), Name: "Default", Slug: "default", CreatedAt: time.Now()}
if _, err := db.Col("orgs").InsertOne(ctx, org); err != nil {
return err
}
server/internal/services/migrate_instance.go
+200
View File
@@ -0,0 +1,200 @@
package services
import (
"context"
"fmt"
"log"
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo"
)
// ScopedCollections lists every collection carrying the tenant key.
//
// Migration 0004 renames org_id to instance_id in each. A collection missing
// from this list keeps the old field name and becomes invisible to every scoped
// query — so this list is load-bearing, not documentation.
//
// AssertNoScopedCollectionMissed checks at boot that nothing outside this list
// holds an org_id.
//
// The migrations collection is deliberately absent: it is not tenant-scoped.
// The two renamed collections appear under their post-rename names, because the
// migration renames the collections before it renames the field.
var ScopedCollections = []string{
"instances",
"servers",
"keys",
"assignments",
"users",
"instance_oidc",
"settings",
"secrets",
"workflows",
"workflow_steps",
"workflow_runs",
"monitors",
"incidents",
"monitor_rollups",
"notification_channels",
"console_sessions",
"audit_logs",
}
// collectionRenames maps the two collections whose names change. Ordered so the
// migration is deterministic.
var collectionRenames = []struct{ from, to string }{
{"orgs", "instances"},
{"org_oidc", "instance_oidc"},
}
// MigrateOrgToInstance renames the tenant key from org_id to instance_id.
//
// It only ever renames documents. It never deletes, drops or unsets one, so a
// bad deploy is recovered by running the inverse rename (cmd/rename-rollback)
// rather than by restoring a backup.
//
// The steps are not atomic across collections — multi-document transactions
// would require a replica set, which self-hosted installs do not guarantee.
// Instead every step is safely repeatable: a collection rename is skipped when
// the source is already gone, and $rename matches nothing on a document that
// has already been renamed. A run that fails partway is fixed by running it
// again.
//
// It must run BEFORE EnsureAuthIndexes. Creating the instances.slug index first
// would create an empty instances collection, and step 1 below refuses to
// rename orgs onto an existing target.
func MigrateOrgToInstance(ctx context.Context, db *mongo.Database) error {
names, err := db.ListCollectionNames(ctx, bson.M{})
if err != nil {
return fmt.Errorf("list collections: %w", err)
}
exists := map[string]bool{}
for _, n := range names {
exists[n] = true
}
// Step 1: rename the collections.
for _, r := range collectionRenames {
switch {
case !exists[r.from]:
// Nothing to rename: either already done or never existed.
continue
case exists[r.to]:
return fmt.Errorf("cannot rename %s to %s: both exist; resolve by hand", r.from, r.to)
}
cmd := bson.D{
{Key: "renameCollection", Value: db.Name() + "." + r.from},
{Key: "to", Value: db.Name() + "." + r.to},
}
if err := db.Client().Database("admin").RunCommand(ctx, cmd).Err(); err != nil {
return fmt.Errorf("rename %s to %s: %w", r.from, r.to, err)
}
log.Printf("0004: renamed collection %s to %s", r.from, r.to)
}
// Step 2: rename the field.
for _, c := range ScopedCollections {
res, err := db.Collection(c).UpdateMany(ctx,
bson.M{"org_id": bson.M{"$exists": true}},
bson.M{"$rename": bson.M{"org_id": "instance_id"}},
)
if err != nil {
return fmt.Errorf("rename org_id in %s: %w", c, err)
}
if res.ModifiedCount > 0 {
log.Printf("0004: %s renamed %d document(s)", c, res.ModifiedCount)
}
}
// Step 3: verify before anyone records a marker. Any mismatch aborts, and
// the migration is re-run rather than marked done.
for _, c := range ScopedCollections {
total, err := db.Collection(c).CountDocuments(ctx, bson.M{})
if err != nil {
return fmt.Errorf("count %s: %w", c, err)
}
if total == 0 {
continue
}
stale, err := db.Collection(c).CountDocuments(ctx, bson.M{"org_id": bson.M{"$exists": true}})
if err != nil {
return fmt.Errorf("count stale in %s: %w", c, err)
}
if stale != 0 {
return fmt.Errorf("%s still has %d document(s) with org_id; migration incomplete", c, stale)
}
scoped, err := db.Collection(c).CountDocuments(ctx, bson.M{"instance_id": bson.M{"$exists": true}})
if err != nil {
return fmt.Errorf("count scoped in %s: %w", c, err)
}
if scoped != total {
return fmt.Errorf("%s has %d document(s) but only %d carry instance_id", c, total, scoped)
}
}
// Step 4: indexes keyed on the old field name now point at a field that no
// longer exists. Drop them; the boot-time index builders recreate the
// current ones. Dropping an index touches no documents.
for _, c := range ScopedCollections {
cur, err := db.Collection(c).Indexes().List(ctx)
if err != nil {
return fmt.Errorf("list indexes on %s: %w", c, err)
}
var specs []bson.M
if err := cur.All(ctx, &specs); err != nil {
return fmt.Errorf("decode indexes on %s: %w", c, err)
}
for _, s := range specs {
name, _ := s["name"].(string)
if name == "_id_" {
continue
}
keys, ok := s["key"].(bson.M)
if !ok {
continue
}
if _, keyed := keys["org_id"]; !keyed {
continue
}
if err := db.Collection(c).Indexes().DropOne(ctx, name); err != nil {
return fmt.Errorf("drop index %s on %s: %w", name, c, err)
}
log.Printf("0004: dropped stale index %s on %s", name, c)
}
}
log.Printf("0004: verified %d collection(s)", len(ScopedCollections))
return nil
}
// AssertNoScopedCollectionMissed reports any collection holding an org_id that
// ScopedCollections does not know about. A hit means a collection was added
// without being added to the list, and its tenant key was never renamed.
func AssertNoScopedCollectionMissed(ctx context.Context, db *mongo.Database) error {
known := map[string]bool{}
for _, c := range ScopedCollections {
known[c] = true
}
names, err := db.ListCollectionNames(ctx, bson.M{})
if err != nil {
return fmt.Errorf("list collections: %w", err)
}
for _, n := range names {
if known[n] {
continue
}
count, err := db.Collection(n).CountDocuments(ctx, bson.M{"org_id": bson.M{"$exists": true}})
if err != nil {
return fmt.Errorf("count %s: %w", n, err)
}
if count > 0 {
return fmt.Errorf("collection %q holds %d document(s) with org_id but is not in ScopedCollections", n, count)
}
}
return nil
}
server/internal/services/monitors.go
+42 -54
View File
@@ -21,7 +21,6 @@ func monCtx() (context.Context, context.CancelFunc) {
return context.WithTimeout(context.Background(), 5*time.Second)
}
func SpecFor(m *models.Monitor) checker.Spec {
return checker.Spec{
Type: m.Type,
@@ -37,10 +36,10 @@ func SpecFor(m *models.Monitor) checker.Spec {
}
}
func ListMonitors(orgID string) ([]models.Monitor, error) {
func ListMonitors(instanceID string) ([]models.Monitor, error) {
ctx, cancel := monCtx()
defer cancel()
cur, err := db.Col("monitors").Find(ctx, bson.M{"org_id": orgID}, options.Find().SetSort(bson.M{"created_at": 1}))
cur, err := db.Col("monitors").Find(ctx, bson.M{"instance_id": instanceID}, options.Find().SetSort(bson.M{"created_at": 1}))
if err != nil {
return nil, err
}
@@ -51,23 +50,23 @@ func ListMonitors(orgID string) ([]models.Monitor, error) {
return out, nil
}
func ListMonitorsForRunner(orgID, runner string) ([]models.Monitor, error) {
if orgID == "" {
func ListMonitorsForRunner(instanceID, runner string) ([]models.Monitor, error) {
if instanceID == "" {
return nil, errors.New("org id required")
}
return listMonitorsForRunner(orgID, runner)
return listMonitorsForRunner(instanceID, runner)
}
func ListServerScheduledMonitors() ([]models.Monitor, error) {
return listMonitorsForRunner("", models.RunnerServer)
}
func listMonitorsForRunner(orgID, runner string) ([]models.Monitor, error) {
func listMonitorsForRunner(instanceID, runner string) ([]models.Monitor, error) {
ctx, cancel := monCtx()
defer cancel()
filter := bson.M{"runner": runner, "enabled": true}
if orgID != "" {
filter["org_id"] = orgID
if instanceID != "" {
filter["instance_id"] = instanceID
}
cur, err := db.Col("monitors").Find(ctx, filter)
if err != nil {
@@ -80,12 +79,11 @@ func listMonitorsForRunner(orgID, runner string) ([]models.Monitor, error) {
return out, nil
}
func GetMonitor(orgID, monitorID string) (*models.Monitor, error) {
func GetMonitor(instanceID, monitorID string) (*models.Monitor, error) {
ctx, cancel := monCtx()
defer cancel()
var m models.Monitor
err := db.Col("monitors").FindOne(ctx, bson.M{"monitor_id": monitorID, "org_id": orgID}).Decode(&m)
err := db.Col("monitors").FindOne(ctx, bson.M{"monitor_id": monitorID, "instance_id": instanceID}).Decode(&m)
if errors.Is(err, mongo.ErrNoDocuments) {
return nil, nil
}
@@ -109,26 +107,26 @@ func getMonitorByID(monitorID string) (*models.Monitor, error) {
return &m, nil
}
func validateRunner(orgID, runner string) error {
func validateRunner(instanceID, runner string) error {
if runner == "" || runner == models.RunnerServer {
return nil
}
if _, err := GetServer(orgID, runner); err != nil {
if _, err := GetServer(instanceID, runner); err != nil {
return fmt.Errorf("runner server %s not found", runner)
}
return nil
}
func CreateMonitor(orgID string, m *models.Monitor) (*models.Monitor, error) {
func CreateMonitor(instanceID string, m *models.Monitor) (*models.Monitor, error) {
ctx, cancel := monCtx()
defer cancel()
if err := validateChannelIDs(orgID, m.ChannelIDs); err != nil {
if err := validateChannelIDs(instanceID, m.ChannelIDs); err != nil {
return nil, err
}
if err := validateRunner(orgID, m.Runner); err != nil {
if err := validateRunner(instanceID, m.Runner); err != nil {
return nil, err
}
m.OrgID = orgID
m.InstanceID = instanceID
m.MonitorID = uuid.NewString()
m.CreatedAt = time.Now()
if m.IntervalSec <= 0 {
@@ -147,17 +145,16 @@ func CreateMonitor(orgID string, m *models.Monitor) (*models.Monitor, error) {
return m, nil
}
func UpdateMonitor(orgID, monitorID string, upd bson.M) error {
func UpdateMonitor(instanceID, monitorID string, upd bson.M) error {
ctx, cancel := monCtx()
defer cancel()
if raw, present := upd["channel_ids"]; present {
ids, ok := raw.([]string)
if !ok {
return fmt.Errorf("channel_ids must be a string array")
}
if err := validateChannelIDs(orgID, ids); err != nil {
if err := validateChannelIDs(instanceID, ids); err != nil {
return err
}
}
@@ -166,43 +163,41 @@ func UpdateMonitor(orgID, monitorID string, upd bson.M) error {
if !ok {
return fmt.Errorf("runner must be a string")
}
if err := validateRunner(orgID, runner); err != nil {
if err := validateRunner(instanceID, runner); err != nil {
return err
}
if runner == "" {
upd["runner"] = models.RunnerServer
}
}
_, err := db.Col("monitors").UpdateOne(ctx, bson.M{"monitor_id": monitorID, "org_id": orgID}, bson.M{"$set": upd})
_, err := db.Col("monitors").UpdateOne(ctx, bson.M{"monitor_id": monitorID, "instance_id": instanceID}, bson.M{"$set": upd})
return err
}
func DeleteMonitor(orgID, monitorID string) error {
func DeleteMonitor(instanceID, monitorID string) error {
ctx, cancel := monCtx()
defer cancel()
res, err := db.Col("monitors").DeleteOne(ctx, bson.M{"monitor_id": monitorID, "org_id": orgID})
res, err := db.Col("monitors").DeleteOne(ctx, bson.M{"monitor_id": monitorID, "instance_id": instanceID})
if err != nil {
return err
}
if res.DeletedCount == 0 {
return nil
}
db.Col("incidents").DeleteMany(ctx, bson.M{"monitor_id": monitorID, "org_id": orgID})
db.Col("monitor_rollups").DeleteMany(ctx, bson.M{"monitor_id": monitorID, "org_id": orgID})
db.Col("incidents").DeleteMany(ctx, bson.M{"monitor_id": monitorID, "instance_id": instanceID})
db.Col("monitor_rollups").DeleteMany(ctx, bson.M{"monitor_id": monitorID, "instance_id": instanceID})
return nil
}
func ListIncidents(orgID, monitorID string, limit int64) ([]models.Incident, error) {
func ListIncidents(instanceID, monitorID string, limit int64) ([]models.Incident, error) {
ctx, cancel := monCtx()
defer cancel()
if limit <= 0 {
limit = 50
}
cur, err := db.Col("incidents").Find(ctx, bson.M{"monitor_id": monitorID, "org_id": orgID},
cur, err := db.Col("incidents").Find(ctx, bson.M{"monitor_id": monitorID, "instance_id": instanceID},
options.Find().SetSort(bson.M{"started_at": -1}).SetLimit(limit))
if err != nil {
return nil, err
@@ -214,12 +209,11 @@ func ListIncidents(orgID, monitorID string, limit int64) ([]models.Incident, err
return out, nil
}
func UptimeRollups(orgID, monitorID string, since time.Time) ([]models.Rollup, error) {
func UptimeRollups(instanceID, monitorID string, since time.Time) ([]models.Rollup, error) {
ctx, cancel := monCtx()
defer cancel()
cur, err := db.Col("monitor_rollups").Find(ctx,
bson.M{"monitor_id": monitorID, "org_id": orgID, "period_start": bson.M{"$gte": since}},
bson.M{"monitor_id": monitorID, "instance_id": instanceID, "period_start": bson.M{"$gte": since}},
options.Find().SetSort(bson.M{"period_start": 1}))
if err != nil {
return nil, err
@@ -231,18 +225,18 @@ func UptimeRollups(orgID, monitorID string, since time.Time) ([]models.Rollup, e
return out, nil
}
func IngestResult(orgID, runner, monitorID string, res checker.Result) error {
if orgID == "" {
func IngestResult(instanceID, runner, monitorID string, res checker.Result) error {
if instanceID == "" {
return errors.New("org id required")
}
return ingestResult(orgID, runner, monitorID, res)
return ingestResult(instanceID, runner, monitorID, res)
}
func IngestServerScheduledResult(monitorID string, res checker.Result) error {
return ingestResult("", models.RunnerServer, monitorID, res)
}
func ingestResult(orgID, runner, monitorID string, res checker.Result) error {
func ingestResult(instanceID, runner, monitorID string, res checker.Result) error {
ctx, cancel := monCtx()
defer cancel()
@@ -253,7 +247,7 @@ func ingestResult(orgID, runner, monitorID string, res checker.Result) error {
if m == nil {
return fmt.Errorf("monitor %s not found", monitorID)
}
if orgID != "" && m.OrgID != orgID {
if instanceID != "" && m.InstanceID != instanceID {
return fmt.Errorf("monitor %s belongs to another org", monitorID)
}
if m.Runner != runner {
@@ -295,28 +289,25 @@ func ingestResult(orgID, runner, monitorID string, res checker.Result) error {
return err
}
bucket := now.Truncate(time.Hour)
up := 0
if res.Up {
up = 1
}
db.Col("monitor_rollups").UpdateOne(ctx,
bson.M{"monitor_id": monitorID, "period_start": bucket},
bson.M{
"$inc": bson.M{"checks": 1, "up_count": up, "sum_latency": int64(res.LatencyMs)},
"$setOnInsert": bson.M{"org_id": m.OrgID},
"$setOnInsert": bson.M{"instance_id": m.InstanceID},
},
options.UpdateOne().SetUpsert(true))
if newStatus != prev {
switch newStatus {
case models.StatusDown:
inc := models.Incident{
OrgID: m.OrgID,
InstanceID: m.InstanceID,
IncidentID: uuid.NewString(),
MonitorID: monitorID,
StartedAt: now,
@@ -327,7 +318,7 @@ func ingestResult(orgID, runner, monitorID string, res checker.Result) error {
case models.StatusUp:
if prev == models.StatusDown {
db.Col("incidents").UpdateOne(ctx,
bson.M{"monitor_id": monitorID, "org_id": m.OrgID, "resolved_at": nil},
bson.M{"monitor_id": monitorID, "instance_id": m.InstanceID, "resolved_at": nil},
bson.M{"$set": bson.M{"resolved_at": now}})
notifyTransition(m, newStatus, res.Message)
}
@@ -336,14 +327,11 @@ func ingestResult(orgID, runner, monitorID string, res checker.Result) error {
return nil
}
func notifyTransition(m *models.Monitor, newStatus, message string) {
if len(m.ChannelIDs) == 0 {
return
}
channels, err := GetChannels(m.OrgID, m.ChannelIDs)
channels, err := GetChannels(m.InstanceID, m.ChannelIDs)
if err != nil {
log.Printf("notify: load channels for %s: %v", m.MonitorID, err)
return
@@ -366,5 +354,5 @@ func notifyTransition(m *models.Monitor, newStatus, message string) {
}
}(ch)
}
_ = UpdateMonitor(m.OrgID, m.MonitorID, bson.M{"state.last_notified_at": time.Now()})
_ = UpdateMonitor(m.InstanceID, m.MonitorID, bson.M{"state.last_notified_at": time.Now()})
}
server/internal/services/secrets.go
+16 -19
View File
@@ -23,7 +23,7 @@ func EnsureSecretIndexes() error {
}
_, err := db.Col("secrets").Indexes().CreateOne(ctx, mongo.IndexModel{
Keys: bson.D{{Key: "org_id", Value: 1}, {Key: "group", Value: 1}, {Key: "key", Value: 1}},
Keys: bson.D{{Key: "instance_id", Value: 1}, {Key: "group", Value: 1}, {Key: "key", Value: 1}},
Options: options.Index().SetUnique(true),
})
return err
@@ -38,12 +38,12 @@ func isIndexNotFound(err error) bool {
return false
}
func ListSecretGroups(orgID string) ([]models.GroupSummary, error) {
func ListSecretGroups(instanceID string) ([]models.GroupSummary, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
pipeline := mongo.Pipeline{
{{Key: "$match", Value: bson.D{{Key: "org_id", Value: orgID}}}},
{{Key: "$match", Value: bson.D{{Key: "instance_id", Value: instanceID}}}},
{{Key: "$group", Value: bson.D{
{Key: "_id", Value: "$group"},
{Key: "key_count", Value: bson.D{{Key: "$sum", Value: 1}}},
@@ -78,11 +78,11 @@ func ListSecretGroups(orgID string) ([]models.GroupSummary, error) {
return groups, nil
}
func GetSecretGroup(orgID, group string) ([]models.Secret, error) {
func GetSecretGroup(instanceID, group string) ([]models.Secret, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
cursor, err := db.Col("secrets").Find(ctx, bson.M{"org_id": orgID, "group": group},
cursor, err := db.Col("secrets").Find(ctx, bson.M{"instance_id": instanceID, "group": group},
options.Find().SetSort(bson.D{{Key: "key", Value: 1}}))
if err != nil {
return nil, err
@@ -96,8 +96,8 @@ func GetSecretGroup(orgID, group string) ([]models.Secret, error) {
return docs, nil
}
func GetSecretGroupDecrypted(orgID, group string) (map[string]string, error) {
docs, err := GetSecretGroup(orgID, group)
func GetSecretGroupDecrypted(instanceID, group string) (map[string]string, error) {
docs, err := GetSecretGroup(instanceID, group)
if err != nil {
return nil, err
}
@@ -112,12 +112,12 @@ func GetSecretGroupDecrypted(orgID, group string) (map[string]string, error) {
return result, nil
}
func RevealSecret(orgID, group, key string) (string, error) {
func RevealSecret(instanceID, group, key string) (string, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
var doc models.Secret
err := db.Col("secrets").FindOne(ctx, bson.M{"org_id": orgID, "group": group, "key": key}).Decode(&doc)
err := db.Col("secrets").FindOne(ctx, bson.M{"instance_id": instanceID, "group": group, "key": key}).Decode(&doc)
if err == mongo.ErrNoDocuments {
return "", fmt.Errorf("secret not found")
}
@@ -127,7 +127,7 @@ func RevealSecret(orgID, group, key string) (string, error) {
return decryptString(doc.EncryptedValue)
}
func UpsertSecrets(orgID, group string, values map[string]string) error {
func UpsertSecrets(instanceID, group string, values map[string]string) error {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
@@ -137,9 +137,9 @@ func UpsertSecrets(orgID, group string, values map[string]string) error {
return fmt.Errorf("encrypt %s: %w", key, err)
}
_, err = db.Col("secrets").UpdateOne(ctx,
bson.M{"org_id": orgID, "group": group, "key": key},
bson.M{"instance_id": instanceID, "group": group, "key": key},
bson.M{"$set": bson.M{
"org_id": orgID,
"instance_id": instanceID,
"encrypted_value": encrypted,
"updated_at": time.Now(),
}},
@@ -152,7 +152,6 @@ func UpsertSecrets(orgID, group string, values map[string]string) error {
return nil
}
func SortedKeys(m map[string]string) []string {
keys := make([]string, 0, len(m))
for k := range m {
@@ -162,20 +161,18 @@ func SortedKeys(m map[string]string) []string {
return keys
}
func DeleteSecret(orgID, group, key string) error {
func DeleteSecret(instanceID, group, key string) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_, err := db.Col("secrets").DeleteOne(ctx, bson.M{"org_id": orgID, "group": group, "key": key})
_, err := db.Col("secrets").DeleteOne(ctx, bson.M{"instance_id": instanceID, "group": group, "key": key})
return err
}
func DeleteSecretGroup(orgID, group string) error {
func DeleteSecretGroup(instanceID, group string) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_, err := db.Col("secrets").DeleteMany(ctx, bson.M{"org_id": orgID, "group": group})
_, err := db.Col("secrets").DeleteMany(ctx, bson.M{"instance_id": instanceID, "group": group})
return err
}
server/internal/services/servers.go
+21 -47
View File
@@ -30,14 +30,14 @@ func HashToken(token string) string {
return hex.EncodeToString(sum[:])
}
func CreateServer(orgID string) (*models.Server, string, error) {
func CreateServer(instanceID string) (*models.Server, string, error) {
token, err := generateToken(32)
if err != nil {
return nil, "", err
}
expires := time.Now().Add(time.Hour)
s := &models.Server{
OrgID: orgID,
InstanceID: instanceID,
ServerID: uuid.NewString(),
PreRegToken: token,
PreRegExpires: &expires,
@@ -54,21 +54,18 @@ func CreateServer(orgID string) (*models.Server, string, error) {
return s, token, nil
}
func GetServer(orgID, serverID string) (*models.Server, error) {
func GetServer(instanceID, serverID string) (*models.Server, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
var s models.Server
err := db.Col("servers").FindOne(ctx, bson.M{"server_id": serverID, "org_id": orgID}).Decode(&s)
err := db.Col("servers").FindOne(ctx, bson.M{"server_id": serverID, "instance_id": instanceID}).Decode(&s)
if err != nil {
return nil, err
}
return &s, nil
}
func getServerByID(serverID string) (*models.Server, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
@@ -96,9 +93,6 @@ func GetServerByPreRegToken(token string) (*models.Server, error) {
return &s, nil
}
func OSTypeFromInfo(osInfo string) string {
if strings.HasPrefix(strings.ToLower(osInfo), "windows") {
return "windows"
@@ -106,8 +100,6 @@ func OSTypeFromInfo(osInfo string) string {
return "linux"
}
func defaultConsoleFields(osType string) (protocols []string, sshPort, rdpPort int) {
if osType == "windows" {
return []string{"rdp"}, 22, 3389
@@ -181,19 +173,13 @@ func ValidateAgentToken(serverID, agentToken string) (*models.Server, error) {
if err != nil {
return nil, fmt.Errorf("invalid agent token")
}
if s.OrgID == "" {
if s.InstanceID == "" {
return nil, fmt.Errorf("server %s has no org", serverID)
}
return &s, nil
}
func BackfillConsoleConfig(srv *models.Server) error {
if srv == nil || len(srv.ConsoleProtocols) > 0 {
return nil
@@ -236,12 +222,12 @@ func UpdateServerLastSeen(serverID, agentVersion string) error {
return err
}
func ListServers(orgID string) ([]models.Server, error) {
func ListServers(instanceID string) ([]models.Server, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
opts := options.Find().SetSort(bson.D{{Key: "created_at", Value: -1}})
cursor, err := db.Col("servers").Find(ctx, bson.M{"org_id": orgID}, opts)
cursor, err := db.Col("servers").Find(ctx, bson.M{"instance_id": instanceID}, opts)
if err != nil {
return nil, err
}
@@ -254,16 +240,16 @@ func ListServers(orgID string) ([]models.Server, error) {
return servers, nil
}
func DeleteServer(orgID, serverID string) error {
func DeleteServer(instanceID, serverID string) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_, err := db.Col("servers").DeleteOne(ctx, bson.M{"server_id": serverID, "org_id": orgID})
_, err := db.Col("servers").DeleteOne(ctx, bson.M{"server_id": serverID, "instance_id": instanceID})
if err != nil {
return err
}
_, err = db.Col("assignments").DeleteMany(ctx, bson.M{"server_id": serverID, "org_id": orgID})
_, err = db.Col("assignments").DeleteMany(ctx, bson.M{"server_id": serverID, "instance_id": instanceID})
return err
}
@@ -283,42 +269,31 @@ func StoreAvailableUpdates(serverID string, pkgs []models.PackageUpdate) error {
}
func MarkOfflineServers() error {
orgIDs, err := ListOrgIDs()
instanceIDs, err := ListInstanceIDs()
if err != nil {
return err
}
for _, orgID := range orgIDs {
if err := markOfflineForFilter(bson.M{"org_id": orgID}, orgID); err != nil {
log.Printf("offline sweep failed for org %s: %v", orgID, err)
for _, instanceID := range instanceIDs {
if err := markOfflineForFilter(bson.M{"instance_id": instanceID}, instanceID); err != nil {
log.Printf("offline sweep failed for org %s: %v", instanceID, err)
}
}
if err := markOfflineForFilter(bson.M{"org_id": bson.M{"$nin": orgIDs}}, ""); err != nil {
if err := markOfflineForFilter(bson.M{"instance_id": bson.M{"$nin": instanceIDs}}, ""); err != nil {
log.Printf("offline sweep failed for orphaned servers: %v", err)
}
return nil
}
func markOfflineForFilter(scope bson.M, orgID string) error {
func markOfflineForFilter(scope bson.M, instanceID string) error {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
var settings *models.Settings
thresholdMinutes := 5
if orgID != "" {
settings, _ = GetSettings(orgID)
if instanceID != "" {
settings, _ = GetSettings(instanceID)
if settings != nil && settings.Alerts.OfflineThresholdMinutes > 0 {
thresholdMinutes = settings.Alerts.OfflineThresholdMinutes
}
@@ -333,7 +308,6 @@ func markOfflineForFilter(scope bson.M, orgID string) error {
filter[k] = v
}
cursor, err := db.Col("servers").Find(ctx, filter)
if err != nil {
return err
@@ -349,7 +323,7 @@ func markOfflineForFilter(scope bson.M, orgID string) error {
}
for _, s := range goingOffline {
LogEvent(s.OrgID, "server.offline", "system", s.ServerID, "", fmt.Sprintf("%s (%s) went offline", s.Hostname, s.IPAddress))
LogEvent(s.InstanceID, "server.offline", "system", s.ServerID, "", fmt.Sprintf("%s (%s) went offline", s.Hostname, s.IPAddress))
if settings != nil && settings.Alerts.Enabled && settings.Alerts.WebhookURL != "" {
go SendOfflineWebhook(settings.Alerts.WebhookURL, s.Hostname, s.ServerID, s.IPAddress)
}
server/internal/services/settings.go
+14 -30
View File
@@ -34,9 +34,6 @@ var defaultSettings = models.Settings{
},
}
func EnsureSettingsIndexes() error {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
@@ -46,16 +43,12 @@ func EnsureSettingsIndexes() error {
}
if _, err := db.Col("settings").Indexes().CreateOne(ctx, mongo.IndexModel{
Keys: bson.D{{Key: "org_id", Value: 1}},
Keys: bson.D{{Key: "instance_id", Value: 1}},
Options: options.Index().SetUnique(true),
}); err != nil {
return err
}
_, err := db.Col("settings").Indexes().CreateOne(ctx, mongo.IndexModel{
Keys: bson.D{{Key: "secrets.read_token_hash", Value: 1}},
Options: options.Index().SetUnique(true).SetName("settings_read_token_hash_unique").
@@ -66,15 +59,15 @@ func EnsureSettingsIndexes() error {
return err
}
func GetSettings(orgID string) (*models.Settings, error) {
func GetSettings(instanceID string) (*models.Settings, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
var s models.Settings
err := db.Col("settings").FindOne(ctx, bson.M{"org_id": orgID}).Decode(&s)
err := db.Col("settings").FindOne(ctx, bson.M{"instance_id": instanceID}).Decode(&s)
if err == mongo.ErrNoDocuments {
cp := defaultSettings
cp.OrgID = orgID
cp.InstanceID = instanceID
return &cp, nil
}
if err != nil {
@@ -89,9 +82,7 @@ func hashToken(token string) string {
return hex.EncodeToString(sum[:])
}
func RotateSecretsReadToken(orgID string) (string, error) {
func RotateSecretsReadToken(instanceID string) (string, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
@@ -102,13 +93,13 @@ func RotateSecretsReadToken(orgID string) (string, error) {
token := hex.EncodeToString(raw)
_, err := db.Col("settings").UpdateOne(ctx,
bson.M{"org_id": orgID},
bson.M{"instance_id": instanceID},
bson.M{
"$set": bson.M{
"secrets.read_token_hash": hashToken(token),
"secrets.rotated_at": time.Now(),
},
"$setOnInsert": bson.M{"org_id": orgID},
"$setOnInsert": bson.M{"instance_id": instanceID},
},
options.UpdateOne().SetUpsert(true),
)
@@ -118,9 +109,6 @@ func RotateSecretsReadToken(orgID string) (string, error) {
return token, nil
}
func ResolveSecretsReadToken(token string) (string, bool) {
if token == "" {
return "", false
@@ -130,7 +118,7 @@ func ResolveSecretsReadToken(token string) (string, bool) {
var s models.Settings
err := db.Col("settings").FindOne(ctx, bson.M{"secrets.read_token_hash": hashToken(token)}).Decode(&s)
if err != nil || s.Secrets.ReadTokenHash == "" || s.OrgID == "" {
if err != nil || s.Secrets.ReadTokenHash == "" || s.InstanceID == "" {
return "", false
}
expected, err := hex.DecodeString(s.Secrets.ReadTokenHash)
@@ -141,10 +129,10 @@ func ResolveSecretsReadToken(token string) (string, bool) {
if subtle.ConstantTimeCompare(expected, got[:]) != 1 {
return "", false
}
return s.OrgID, true
return s.InstanceID, true
}
func SaveSettings(orgID string, alerts models.AlertSettings, email models.EmailSettings, retentionDays *int) error {
func SaveSettings(instanceID string, alerts models.AlertSettings, email models.EmailSettings, retentionDays *int) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
@@ -160,17 +148,15 @@ func SaveSettings(orgID string, alerts models.AlertSettings, email models.EmailS
set["workflow_log_retention_days"] = *retentionDays
}
_, err := db.Col("settings").UpdateOne(ctx,
bson.M{"org_id": orgID},
bson.M{"$set": set, "$setOnInsert": bson.M{"org_id": orgID}},
bson.M{"instance_id": instanceID},
bson.M{"$set": set, "$setOnInsert": bson.M{"instance_id": instanceID}},
options.UpdateOne().SetUpsert(true),
)
return err
}
func GetWorkflowLogRetentionDays(orgID string) (int, error) {
s, err := GetSettings(orgID)
func GetWorkflowLogRetentionDays(instanceID string) (int, error) {
s, err := GetSettings(instanceID)
if err != nil {
return 30, err
}
@@ -241,7 +227,6 @@ func SendOfflineEmail(cfg models.EmailSettings, hostname, serverID, ipAddress st
}
}
func sendMailTLS(addr, host string, auth smtp.Auth, from string, to []string, msg []byte) error {
conn, err := tls.Dial("tcp", addr, &tls.Config{ServerName: host})
if err != nil {
@@ -278,4 +263,3 @@ func sendMailTLS(addr, host string, auth smtp.Auth, from string, to []string, ms
}
return c.Quit()
}
server/internal/services/stepio.go
+4 -10
View File
@@ -9,7 +9,6 @@ import (
const StepDocKind = "vantage.step/v1"
type StepDoc struct {
Kind string `json:"kind"`
Name string `json:"name"`
@@ -21,7 +20,6 @@ type StepDoc struct {
SecretRefs []string `json:"secret_refs"`
}
func ExportStepDoc(s models.WorkflowStep) StepDoc {
return StepDoc{
Kind: StepDocKind,
@@ -35,8 +33,6 @@ func ExportStepDoc(s models.WorkflowStep) StepDoc {
}
}
func ParseStepDoc(b []byte) (models.WorkflowStep, error) {
var d StepDoc
if err := json.Unmarshal(b, &d); err != nil {
@@ -65,20 +61,18 @@ func ParseStepDoc(b []byte) (models.WorkflowStep, error) {
}, nil
}
func ImportStepToLibrary(orgID string, b []byte) (*models.WorkflowStep, error) {
func ImportStepToLibrary(instanceID string, b []byte) (*models.WorkflowStep, error) {
s, err := ParseStepDoc(b)
if err != nil {
return nil, err
}
return CreateStep(orgID, s)
return CreateStep(instanceID, s)
}
func ExportStep(orgID, stepID string) ([]byte, error) {
func ExportStep(instanceID, stepID string) ([]byte, error) {
ctx, cancel := wfCtx()
defer cancel()
s, err := getStep(ctx, orgID, stepID)
s, err := getStep(ctx, instanceID, stepID)
if err != nil {
return nil, err
}
server/internal/services/steplogs.go
+13 -40
View File
@@ -14,7 +14,6 @@ import (
"go.mongodb.org/mongo-driver/v2/mongo"
)
func WorkflowLogDir() string {
dir := os.Getenv("VANTAGE_WORKFLOW_LOG_DIR")
if dir == "" {
@@ -24,19 +23,14 @@ func WorkflowLogDir() string {
return dir
}
func ServerRunLogPath(runID, serverID string) string {
return filepath.Join(WorkflowLogDir(), runID, serverID+".log")
}
func logTS() string {
return time.Now().UTC().Format("2006-01-02T15:04:05.000") + "Z"
}
func AppendMarker(runID, serverID, text string) (int64, error) {
path := ServerRunLogPath(runID, serverID)
if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {
@@ -47,19 +41,17 @@ func AppendMarker(runID, serverID, text string) (int64, error) {
return 0, err
}
defer f.Close()
off, _ := f.Seek(0, 2)
off, _ := f.Seek(0, 2)
if _, err := f.WriteString("[" + logTS() + "] " + text + "\n"); err != nil {
return off, err
}
return off, nil
}
type stepLogWriter struct {
mu sync.Mutex
f *os.File
carry []byte
carry []byte
secrets []string
}
@@ -70,7 +62,6 @@ type stepLogRegistry struct {
var StepLogs = &stepLogRegistry{writers: make(map[string]*stepLogWriter)}
func (r *stepLogRegistry) Open(commandID, path string, secrets []string) error {
if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {
return err
@@ -92,10 +83,6 @@ func (r *stepLogRegistry) get(commandID string) *stepLogWriter {
return r.writers[commandID]
}
func (r *stepLogRegistry) Append(commandID string, data []byte) {
w := r.get(commandID)
if w == nil {
@@ -115,7 +102,6 @@ func (r *stepLogRegistry) Append(commandID string, data []byte) {
w.carry = append([]byte{}, buf...)
}
func (w *stepLogWriter) writeLine(line []byte) {
masked := maskBytes(line, w.secrets)
_, _ = w.f.WriteString("[" + logTS() + "] ")
@@ -123,7 +109,6 @@ func (w *stepLogWriter) writeLine(line []byte) {
_, _ = w.f.WriteString("\n")
}
func (r *stepLogRegistry) Close(commandID string) {
r.mu.Lock()
w := r.writers[commandID]
@@ -152,9 +137,6 @@ func maskBytes(b []byte, secrets []string) []byte {
return []byte(s)
}
func StartLogSweeper() {
go func() {
sweepLogs()
@@ -166,10 +148,6 @@ func StartLogSweeper() {
}()
}
func sweepLogs() {
base := WorkflowLogDir()
entries, err := os.ReadDir(base)
@@ -186,30 +164,28 @@ func sweepLogs() {
runID := e.Name()
dir := filepath.Join(base, runID)
orgID, finishedAt, found, err := runRetentionInfo(runID)
instanceID, finishedAt, found, err := runRetentionInfo(runID)
if err != nil {
log.Printf("log sweep: retention lookup failed for run %s: %v", runID, err)
continue
}
if found && finishedAt == nil {
continue
continue
}
days, ok := cache[orgID]
days, ok := cache[instanceID]
if !ok {
days = defaultRetentionDays
if orgID != "" {
if v, err := GetWorkflowLogRetentionDays(orgID); err == nil {
if instanceID != "" {
if v, err := GetWorkflowLogRetentionDays(instanceID); err == nil {
days = v
}
}
cache[orgID] = days
cache[instanceID] = days
}
if days <= 0 {
continue
continue
}
cutoff := now.AddDate(0, 0, -days)
@@ -219,7 +195,7 @@ func sweepLogs() {
}
continue
}
if fi, e := os.Stat(dir); e == nil && fi.ModTime().Before(cutoff) {
_ = os.RemoveAll(dir)
}
@@ -228,14 +204,11 @@ func sweepLogs() {
const defaultRetentionDays = 30
func runRetentionInfo(runID string) (string, *time.Time, bool, error) {
ctx, cancel := wfCtx()
defer cancel()
var run struct {
OrgID string `bson:"org_id"`
InstanceID string `bson:"instance_id"`
FinishedAt *time.Time `bson:"finished_at"`
}
err := db.Col("workflow_runs").FindOne(ctx, bson.M{"run_id": runID}).Decode(&run)
@@ -245,5 +218,5 @@ func runRetentionInfo(runID string) (string, *time.Time, bool, error) {
if err != nil {
return "", nil, false, err
}
return run.OrgID, run.FinishedAt, true, nil
return run.InstanceID, run.FinishedAt, true, nil
}
server/internal/services/stepresults.go
-6
View File
@@ -11,12 +11,8 @@ type stepResultRegistry struct {
pending map[string]chan *pb.StepResult
}
var StepResults = &stepResultRegistry{pending: make(map[string]chan *pb.StepResult)}
func (r *stepResultRegistry) Await(commandID string) <-chan *pb.StepResult {
ch := make(chan *pb.StepResult, 1)
r.mu.Lock()
@@ -25,14 +21,12 @@ func (r *stepResultRegistry) Await(commandID string) <-chan *pb.StepResult {
return ch
}
func (r *stepResultRegistry) Cancel(commandID string) {
r.mu.Lock()
delete(r.pending, commandID)
r.mu.Unlock()
}
func (r *stepResultRegistry) Deliver(res *pb.StepResult) {
if res == nil {
return
server/internal/services/stepscan.go
+1 -6
View File
@@ -5,12 +5,8 @@ import (
"strings"
)
var keyAssign = regexp.MustCompile(`([A-Za-z_][A-Za-z0-9_]*)=`)
func DeriveOutputs(script string) []string {
out := []string{}
seen := map[string]bool{}
@@ -20,7 +16,7 @@ func DeriveOutputs(script string) []string {
}
for _, m := range keyAssign.FindAllStringSubmatch(line, -1) {
key := m[1]
if key == "WORKFLOW_ENV" || key == "env" {
continue
}
@@ -36,4 +32,3 @@ func DeriveOutputs(script string) []string {
// Slugify lived here and was mirrored by hand in sitesvc. It now has a single
// definition in shared/provision, which both services import.
server/internal/services/sync.go
+4 -6
View File
@@ -13,17 +13,15 @@ func BuildAuthorizedKeys(serverID string) ([]string, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
srv, err := getServerByID(serverID)
if err != nil {
return nil, err
}
cursor, err := db.Col("assignments").Find(ctx, bson.M{
"org_id": srv.OrgID,
"server_id": serverID,
"revoked_at": nil,
"instance_id": srv.InstanceID,
"server_id": serverID,
"revoked_at": nil,
})
if err != nil {
return nil, err
@@ -38,7 +36,7 @@ func BuildAuthorizedKeys(serverID string) ([]string, error) {
var lines []string
for _, a := range assignments {
var key models.Key
err := db.Col("keys").FindOne(ctx, bson.M{"key_id": a.KeyID, "org_id": srv.OrgID}).Decode(&key)
err := db.Col("keys").FindOne(ctx, bson.M{"key_id": a.KeyID, "instance_id": srv.InstanceID}).Decode(&key)
if err != nil {
continue
}
server/internal/services/users.go
+21 -28
View File
@@ -14,53 +14,46 @@ import (
"golang.org/x/crypto/bcrypt"
)
var ErrLastOwner = errors.New("this is the organization's last owner promote another member to owner first")
func CountUsers() (int64, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
return db.Col("users").CountDocuments(ctx, bson.M{})
}
func CountOrgUsers(orgID string) (int64, error) {
func CountInstanceUsers(instanceID string) (int64, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
return db.Col("users").CountDocuments(ctx, bson.M{"org_id": orgID})
return db.Col("users").CountDocuments(ctx, bson.M{"instance_id": instanceID})
}
func countOtherOwners(orgID, exceptUserID string) (int64, error) {
func countOtherOwners(instanceID, exceptUserID string) (int64, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
return db.Col("users").CountDocuments(ctx, bson.M{
"org_id": orgID,
"role": models.RoleOwner,
"user_id": bson.M{"$ne": exceptUserID},
"instance_id": instanceID,
"role": models.RoleOwner,
"user_id": bson.M{"$ne": exceptUserID},
})
}
func GetUserInOrg(orgID, userID string) (*models.User, error) {
func GetUserInInstance(instanceID, userID string) (*models.User, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
var u models.User
err := db.Col("users").FindOne(ctx, bson.M{"user_id": userID, "org_id": orgID}).Decode(&u)
err := db.Col("users").FindOne(ctx, bson.M{"user_id": userID, "instance_id": instanceID}).Decode(&u)
if err != nil {
return nil, err
}
return &u, nil
}
func CreateUser(orgID, email, password, role, authSource string) (*models.User, error) {
func CreateUser(instanceID, email, password, role, authSource string) (*models.User, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
u, err := provision.CreateUser(ctx, db.Database, orgID, email, password, role, authSource)
u, err := provision.CreateUser(ctx, db.Database, instanceID, email, password, role, authSource)
if errors.Is(err, provision.ErrEmailTaken) {
// Preserve the exact error string the API returned before this call
// was delegated to the shared module.
@@ -97,10 +90,10 @@ func TouchLastLogin(userID string) error {
return err
}
func ListUsers(orgID string) ([]models.User, error) {
func ListUsers(instanceID string) ([]models.User, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
cursor, err := db.Col("users").Find(ctx, bson.M{"org_id": orgID})
cursor, err := db.Col("users").Find(ctx, bson.M{"instance_id": instanceID})
if err != nil {
return nil, err
}
@@ -112,17 +105,17 @@ func ListUsers(orgID string) ([]models.User, error) {
return users, nil
}
func UpdateUserRole(orgID, userID, role string) error {
func UpdateUserRole(instanceID, userID, role string) error {
if !models.ValidRole(role) {
return fmt.Errorf("invalid role %q", role)
}
target, err := GetUserInOrg(orgID, userID)
target, err := GetUserInInstance(instanceID, userID)
if err != nil {
return fmt.Errorf("user not found")
}
if target.Role == models.RoleOwner && role != models.RoleOwner {
others, err := countOtherOwners(orgID, userID)
others, err := countOtherOwners(instanceID, userID)
if err != nil {
return err
}
@@ -134,18 +127,18 @@ func UpdateUserRole(orgID, userID, role string) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_, err = db.Col("users").UpdateOne(ctx,
bson.M{"user_id": userID, "org_id": orgID},
bson.M{"user_id": userID, "instance_id": instanceID},
bson.M{"$set": bson.M{"role": role}})
return err
}
func DeleteUser(orgID, userID string) error {
target, err := GetUserInOrg(orgID, userID)
func DeleteUser(instanceID, userID string) error {
target, err := GetUserInInstance(instanceID, userID)
if err != nil {
return fmt.Errorf("user not found")
}
if target.Role == models.RoleOwner {
others, err := countOtherOwners(orgID, userID)
others, err := countOtherOwners(instanceID, userID)
if err != nil {
return err
}
@@ -156,6 +149,6 @@ func DeleteUser(orgID, userID string) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_, err = db.Col("users").DeleteOne(ctx, bson.M{"user_id": userID, "org_id": orgID})
_, err = db.Col("users").DeleteOne(ctx, bson.M{"user_id": userID, "instance_id": instanceID})
return err
}
server/internal/services/validate.go
-1
View File
@@ -6,7 +6,6 @@ import (
"github.com/mrhid6/vantage/server/internal/models"
)
func ValidateWorkflow(w models.Workflow) error {
for i, ref := range w.Steps {
hasLib := ref.StepID != ""
server/internal/services/workflow_runner.go
+19 -19
View File
@@ -17,8 +17,8 @@ import (
const stepDispatchGrace = 15 * time.Second
func TriggerWorkflow(orgID, workflowID, actor string) (string, error) {
wf, err := GetWorkflow(orgID, workflowID)
func TriggerWorkflow(instanceID, workflowID, actor string) (string, error) {
wf, err := GetWorkflow(instanceID, workflowID)
if err != nil {
return "", err
}
@@ -29,24 +29,24 @@ func TriggerWorkflow(orgID, workflowID, actor string) (string, error) {
return "", fmt.Errorf("workflow has no steps")
}
if err := validateTargetServers(orgID, wf.TargetServerIDs); err != nil {
if err := validateTargetServers(instanceID, wf.TargetServerIDs); err != nil {
return "", err
}
ctx, cancel := wfCtx()
running := db.Col("workflow_runs").FindOne(ctx, bson.M{"org_id": orgID, "workflow_id": workflowID, "status": "running"})
running := db.Col("workflow_runs").FindOne(ctx, bson.M{"instance_id": instanceID, "workflow_id": workflowID, "status": "running"})
cancel()
if running.Err() == nil {
return "", fmt.Errorf("workflow already has a run in progress")
}
resolved, err := resolveSteps(orgID, wf)
resolved, err := resolveSteps(instanceID, wf)
if err != nil {
return "", err
}
run := models.WorkflowRun{
OrgID: orgID,
InstanceID: instanceID,
RunID: uuid.New().String(),
WorkflowID: workflowID,
Name: wf.Name,
@@ -78,7 +78,7 @@ func TriggerWorkflow(orgID, workflowID, actor string) (string, error) {
return run.RunID, nil
}
func resolveSteps(orgID string, wf *models.Workflow) ([]models.ResolvedStep, error) {
func resolveSteps(instanceID string, wf *models.Workflow) ([]models.ResolvedStep, error) {
ctx, cancel := wfCtx()
defer cancel()
out := make([]models.ResolvedStep, 0, len(wf.Steps))
@@ -87,7 +87,7 @@ func resolveSteps(orgID string, wf *models.Workflow) ([]models.ResolvedStep, err
out = append(out, resolveInlineStep(ref))
continue
}
lib, err := getStep(ctx, orgID, ref.StepID)
lib, err := getStep(ctx, instanceID, ref.StepID)
if err != nil {
return nil, err
}
@@ -163,7 +163,7 @@ func executeRun(runID string) {
done := make(chan int, len(run.ServerRuns))
for i := range run.ServerRuns {
go func(idx int) {
runServer(run.OrgID, runID, idx, run.Steps, run.ServerRuns[idx].ServerID)
runServer(run.InstanceID, runID, idx, run.Steps, run.ServerRuns[idx].ServerID)
done <- idx
}(i)
}
@@ -185,7 +185,7 @@ func executeRun(runID string) {
bson.M{"$set": bson.M{"status": status, "finished_at": now}})
}
func runServer(orgID, runID string, srvIdx int, steps []models.ResolvedStep, serverID string) {
func runServer(instanceID, runID string, srvIdx int, steps []models.ResolvedStep, serverID string) {
now := time.Now()
setServerRun(runID, srvIdx, bson.M{"server_runs.$.status": "running", "server_runs.$.started_at": now})
@@ -212,7 +212,7 @@ func runServer(orgID, runID string, srvIdx int, steps []models.ResolvedStep, ser
maxAttempts = step.MaxRetries + 1
}
secretVals := resolveSecrets(orgID, step.SecretRefs)
secretVals := resolveSecrets(instanceID, step.SecretRefs)
for k, v := range secretVals {
allSecrets[k] = v
}
@@ -347,7 +347,7 @@ func expandVars(v string, lookup map[string]string) string {
})
}
func resolveSecrets(orgID string, refs []string) map[string]string {
func resolveSecrets(instanceID string, refs []string) map[string]string {
out := map[string]string{}
for _, ref := range refs {
@@ -355,7 +355,7 @@ func resolveSecrets(orgID string, refs []string) map[string]string {
if len(parts) != 2 {
continue
}
if v, err := RevealSecret(orgID, parts[0], parts[1]); err == nil {
if v, err := RevealSecret(instanceID, parts[0], parts[1]); err == nil {
out[parts[1]] = v
}
}
@@ -453,21 +453,21 @@ func getRunByID(runID string) (*models.WorkflowRun, error) {
return &r, err
}
func GetRun(orgID, runID string) (*models.WorkflowRun, error) {
func GetRun(instanceID, runID string) (*models.WorkflowRun, error) {
ctx, cancel := wfCtx()
defer cancel()
var r models.WorkflowRun
err := db.Col("workflow_runs").FindOne(ctx, bson.M{"run_id": runID, "org_id": orgID}).Decode(&r)
err := db.Col("workflow_runs").FindOne(ctx, bson.M{"run_id": runID, "instance_id": instanceID}).Decode(&r)
if err == mongo.ErrNoDocuments {
return nil, fmt.Errorf("run not found")
}
return &r, err
}
func ListRuns(orgID, workflowID string, limit int64) ([]models.WorkflowRun, error) {
func ListRuns(instanceID, workflowID string, limit int64) ([]models.WorkflowRun, error) {
ctx, cancel := wfCtx()
defer cancel()
cur, err := db.Col("workflow_runs").Find(ctx, bson.M{"org_id": orgID, "workflow_id": workflowID},
cur, err := db.Col("workflow_runs").Find(ctx, bson.M{"instance_id": instanceID, "workflow_id": workflowID},
options.Find().SetSort(bson.D{{Key: "started_at", Value: -1}}).SetLimit(limit))
if err != nil {
return nil, err
@@ -480,12 +480,12 @@ func ListRuns(orgID, workflowID string, limit int64) ([]models.WorkflowRun, erro
return runs, nil
}
func CancelRun(orgID, runID string) error {
func CancelRun(instanceID, runID string) error {
now := time.Now()
ctx, cancel := wfCtx()
defer cancel()
_, err := db.Col("workflow_runs").UpdateOne(ctx,
bson.M{"org_id": orgID, "run_id": runID, "status": "running"},
bson.M{"instance_id": instanceID, "run_id": runID, "status": "running"},
bson.M{"$set": bson.M{"status": "cancelled", "finished_at": now}})
return err
}
server/internal/services/workflows.go
+30 -42
View File
@@ -25,13 +25,12 @@ func EnsureWorkflowIndexes() error {
}); err != nil {
return err
}
if err := db.Col("workflow_steps").Indexes().DropOne(ctx, "slug_1"); err != nil && !isIndexNotFound(err) {
return err
}
if _, err := db.Col("workflow_steps").Indexes().CreateOne(ctx, mongo.IndexModel{
Keys: bson.D{{Key: "org_id", Value: 1}, {Key: "slug", Value: 1}},
Keys: bson.D{{Key: "instance_id", Value: 1}, {Key: "slug", Value: 1}},
Options: options.Index().SetUnique(true).
SetPartialFilterExpression(bson.M{"source": "default"}),
}); err != nil {
@@ -48,12 +47,10 @@ func EnsureWorkflowIndexes() error {
return err
}
func ListSteps(orgID string) ([]models.WorkflowStep, error) {
func ListSteps(instanceID string) ([]models.WorkflowStep, error) {
ctx, cancel := wfCtx()
defer cancel()
cur, err := db.Col("workflow_steps").Find(ctx, bson.M{"org_id": orgID},
cur, err := db.Col("workflow_steps").Find(ctx, bson.M{"instance_id": instanceID},
options.Find().SetSort(bson.D{{Key: "name", Value: 1}}))
if err != nil {
return nil, err
@@ -66,12 +63,10 @@ func ListSteps(orgID string) ([]models.WorkflowStep, error) {
return steps, nil
}
func StepUsageCounts(orgID string) (map[string]int, error) {
func StepUsageCounts(instanceID string) (map[string]int, error) {
ctx, cancel := wfCtx()
defer cancel()
cur, err := db.Col("workflows").Find(ctx, bson.M{"org_id": orgID})
cur, err := db.Col("workflows").Find(ctx, bson.M{"instance_id": instanceID})
if err != nil {
return nil, err
}
@@ -94,10 +89,10 @@ func StepUsageCounts(orgID string) (map[string]int, error) {
return counts, nil
}
func CreateStep(orgID string, s models.WorkflowStep) (*models.WorkflowStep, error) {
func CreateStep(instanceID string, s models.WorkflowStep) (*models.WorkflowStep, error) {
ctx, cancel := wfCtx()
defer cancel()
s.OrgID = orgID
s.InstanceID = instanceID
s.StepID = uuid.New().String()
s.CreatedAt = time.Now()
s.UpdatedAt = s.CreatedAt
@@ -117,10 +112,10 @@ func CreateStep(orgID string, s models.WorkflowStep) (*models.WorkflowStep, erro
return &s, nil
}
func UpdateStep(orgID, stepID string, s models.WorkflowStep) error {
func UpdateStep(instanceID, stepID string, s models.WorkflowStep) error {
ctx, cancel := wfCtx()
defer cancel()
_, err := db.Col("workflow_steps").UpdateOne(ctx, bson.M{"step_id": stepID, "org_id": orgID}, bson.M{"$set": bson.M{
_, err := db.Col("workflow_steps").UpdateOne(ctx, bson.M{"step_id": stepID, "instance_id": instanceID}, bson.M{"$set": bson.M{
"name": s.Name,
"description": s.Description,
"interpreter": s.Interpreter,
@@ -133,14 +128,14 @@ func UpdateStep(orgID, stepID string, s models.WorkflowStep) error {
return err
}
func DeleteStep(orgID, stepID string) error {
func DeleteStep(instanceID, stepID string) error {
ctx, cancel := wfCtx()
defer cancel()
if _, err := db.Col("workflow_steps").DeleteOne(ctx, bson.M{"step_id": stepID, "org_id": orgID}); err != nil {
if _, err := db.Col("workflow_steps").DeleteOne(ctx, bson.M{"step_id": stepID, "instance_id": instanceID}); err != nil {
return err
}
cur, err := db.Col("workflows").Find(ctx, bson.M{"steps.step_id": stepID, "org_id": orgID})
cur, err := db.Col("workflows").Find(ctx, bson.M{"steps.step_id": stepID, "instance_id": instanceID})
if err != nil {
return err
}
@@ -170,21 +165,19 @@ func DeleteStep(orgID, stepID string) error {
return nil
}
func getStep(ctx context.Context, orgID, stepID string) (*models.WorkflowStep, error) {
func getStep(ctx context.Context, instanceID, stepID string) (*models.WorkflowStep, error) {
var s models.WorkflowStep
err := db.Col("workflow_steps").FindOne(ctx, bson.M{"step_id": stepID, "org_id": orgID}).Decode(&s)
err := db.Col("workflow_steps").FindOne(ctx, bson.M{"step_id": stepID, "instance_id": instanceID}).Decode(&s)
if err == mongo.ErrNoDocuments {
return nil, fmt.Errorf("step %s not found", stepID)
}
return &s, err
}
func ListWorkflows(orgID string) ([]models.Workflow, error) {
func ListWorkflows(instanceID string) ([]models.Workflow, error) {
ctx, cancel := wfCtx()
defer cancel()
cur, err := db.Col("workflows").Find(ctx, bson.M{"org_id": orgID},
cur, err := db.Col("workflows").Find(ctx, bson.M{"instance_id": instanceID},
options.Find().SetSort(bson.D{{Key: "name", Value: 1}}))
if err != nil {
return nil, err
@@ -197,21 +190,21 @@ func ListWorkflows(orgID string) ([]models.Workflow, error) {
return wfs, nil
}
func GetWorkflow(orgID, id string) (*models.Workflow, error) {
func GetWorkflow(instanceID, id string) (*models.Workflow, error) {
ctx, cancel := wfCtx()
defer cancel()
var w models.Workflow
err := db.Col("workflows").FindOne(ctx, bson.M{"workflow_id": id, "org_id": orgID}).Decode(&w)
err := db.Col("workflows").FindOne(ctx, bson.M{"workflow_id": id, "instance_id": instanceID}).Decode(&w)
if err == mongo.ErrNoDocuments {
return nil, fmt.Errorf("workflow not found")
}
return &w, err
}
func CreateWorkflow(orgID string, w models.Workflow) (*models.Workflow, error) {
func CreateWorkflow(instanceID string, w models.Workflow) (*models.Workflow, error) {
ctx, cancel := wfCtx()
defer cancel()
w.OrgID = orgID
w.InstanceID = instanceID
w.WorkflowID = uuid.New().String()
w.CreatedAt = time.Now()
w.UpdatedAt = w.CreatedAt
@@ -224,7 +217,7 @@ func CreateWorkflow(orgID string, w models.Workflow) (*models.Workflow, error) {
if err := ValidateWorkflow(w); err != nil {
return nil, err
}
if err := validateTargetServers(orgID, w.TargetServerIDs); err != nil {
if err := validateTargetServers(instanceID, w.TargetServerIDs); err != nil {
return nil, err
}
normalizeInlineSteps(&w)
@@ -234,17 +227,17 @@ func CreateWorkflow(orgID string, w models.Workflow) (*models.Workflow, error) {
return &w, nil
}
func UpdateWorkflow(orgID, id string, w models.Workflow) error {
func UpdateWorkflow(instanceID, id string, w models.Workflow) error {
ctx, cancel := wfCtx()
defer cancel()
if err := ValidateWorkflow(w); err != nil {
return err
}
if err := validateTargetServers(orgID, w.TargetServerIDs); err != nil {
if err := validateTargetServers(instanceID, w.TargetServerIDs); err != nil {
return err
}
normalizeInlineSteps(&w)
_, err := db.Col("workflows").UpdateOne(ctx, bson.M{"workflow_id": id, "org_id": orgID}, bson.M{"$set": bson.M{
_, err := db.Col("workflows").UpdateOne(ctx, bson.M{"workflow_id": id, "instance_id": instanceID}, bson.M{"$set": bson.M{
"name": w.Name,
"target_server_ids": w.TargetServerIDs,
"steps": w.Steps,
@@ -253,20 +246,15 @@ func UpdateWorkflow(orgID, id string, w models.Workflow) error {
return err
}
func validateTargetServers(orgID string, serverIDs []string) error {
func validateTargetServers(instanceID string, serverIDs []string) error {
for _, sid := range serverIDs {
if _, err := GetServer(orgID, sid); err != nil {
if _, err := GetServer(instanceID, sid); err != nil {
return fmt.Errorf("target server %s not found", sid)
}
}
return nil
}
func normalizeInlineSteps(w *models.Workflow) {
for i := range w.Steps {
in := w.Steps[i].Inline
@@ -288,9 +276,9 @@ func normalizeInlineSteps(w *models.Workflow) {
}
}
func DeleteWorkflow(orgID, id string) error {
func DeleteWorkflow(instanceID, id string) error {
ctx, cancel := wfCtx()
defer cancel()
_, err := db.Col("workflows").DeleteOne(ctx, bson.M{"workflow_id": id, "org_id": orgID})
_, err := db.Col("workflows").DeleteOne(ctx, bson.M{"workflow_id": id, "instance_id": instanceID})
return err
}