From 3c1d2201ce684bb8c89d4eaf17b9fc86368a2090 Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Tue, 19 Nov 2024 19:35:38 +1000 Subject: [PATCH] add methods to create, update, list, delete containers in store --- internal/machine/store/container.go | 113 ++++++++++++++++++++++++++++ internal/machine/store/store.go | 31 -------- 2 files changed, 113 insertions(+), 31 deletions(-) diff --git a/internal/machine/store/container.go b/internal/machine/store/container.go index c9de5852..4c2b6370 100644 --- a/internal/machine/store/container.go +++ b/internal/machine/store/container.go @@ -1,6 +1,11 @@ package store import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "strings" "time" "uncloud/internal/machine/docker/container" ) @@ -21,3 +26,111 @@ type ContainerRecord struct { SyncStatus string UpdatedAt time.Time } + +type ListOptions struct { + // MachineIDs filters containers by the machine IDs they are running on. + MachineIDs []string +} + +type DeleteOptions struct { + IDs []string +} + +// CreateOrUpdateContainer creates a new container record or updates an existing one in the store database. +// The container is associated with the given machine ID that indicates which machine the container is running on. +func (s *Store) CreateOrUpdateContainer(ctx context.Context, c *container.Container, machineID string) error { + cJSON, err := json.Marshal(c) + if err != nil { + return fmt.Errorf("marshal container: %w", err) + } + + // Insert or update the container record if the container or machine ID has changed. + res, err := s.corro.ExecContext(ctx, ` + INSERT INTO containers (id, container, machine_id, sync_status, updated_at) + VALUES (?, ?, ?, ?, datetime('now')) + ON CONFLICT (id) DO UPDATE SET container = excluded.container, + machine_id = excluded.machine_id, + sync_status = excluded.sync_status, + updated_at = excluded.updated_at + WHERE containers.container != excluded.container + OR containers.machine_id != excluded.machine_id`, + c.ID, string(cJSON), machineID, SyncStatusSynced) + if err != nil { + return fmt.Errorf("upsert query: %w", err) + } + if res.RowsAffected > 0 { + slog.Debug("Container record updated in store DB.", "id", c.ID, "machine_id", machineID) + } + + return nil +} + +// ListContainers returns a list of container records from the store database that match the given options. +func (s *Store) ListContainers(ctx context.Context, opts ListOptions) ([]*ContainerRecord, error) { + query := "SELECT container, machine_id, sync_status, updated_at FROM containers" + var args []any + + if len(opts.MachineIDs) > 0 { + query += " WHERE machine_id IN (?" + strings.Repeat(", ?", len(opts.MachineIDs)-1) + ")" + args = make([]any, len(opts.MachineIDs)) + for i, id := range opts.MachineIDs { + args[i] = id + } + } + + rows, err := s.corro.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("select query: %w", err) + } + defer rows.Close() + + var containers []*ContainerRecord + var cJSON, machineID, syncStatus, updatedAtStr string + var updatedAt time.Time + + for rows.Next() { + if err = rows.Scan(&cJSON, &machineID, &syncStatus, &updatedAtStr); err != nil { + return nil, fmt.Errorf("scan container record: %w", err) + } + + var c container.Container + if err = json.Unmarshal([]byte(cJSON), &c); err != nil { + return nil, fmt.Errorf("unmarshal container: %w", err) + } + if updatedAt, err = time.Parse(time.DateTime, updatedAtStr); err != nil { + return nil, fmt.Errorf("parse updated_at: %w", err) + } + containers = append(containers, &ContainerRecord{ + Container: &c, + MachineID: machineID, + SyncStatus: syncStatus, + UpdatedAt: updatedAt, + }) + } + + return containers, nil +} + +// DeleteContainers deletes container records from the store database that match the given options. +func (s *Store) DeleteContainers(ctx context.Context, opts DeleteOptions) error { + query := "DELETE FROM containers" + var args []any + + if len(opts.IDs) > 0 { + query += " WHERE id IN (?" + strings.Repeat(", ?", len(opts.IDs)-1) + ")" + args = make([]any, len(opts.IDs)) + for i, id := range opts.IDs { + args[i] = id + } + } + + res, err := s.corro.ExecContext(ctx, query, args...) + if err != nil { + return fmt.Errorf("delete query: %w", err) + } + if res.RowsAffected > 0 { + slog.Debug("Container records deleted from store DB.", "ids", opts.IDs, "count", res.RowsAffected) + } + + return nil +} diff --git a/internal/machine/store/store.go b/internal/machine/store/store.go index 7166b736..8bb7ff30 100644 --- a/internal/machine/store/store.go +++ b/internal/machine/store/store.go @@ -3,14 +3,12 @@ package store import ( "context" _ "embed" - "encoding/json" "errors" "fmt" "google.golang.org/protobuf/encoding/protojson" "log/slog" "uncloud/internal/corrosion" "uncloud/internal/machine/api/pb" - "uncloud/internal/machine/docker/container" ) var ( @@ -131,32 +129,3 @@ func (s *Store) SubscribeMachines(ctx context.Context) ([]*pb.MachineInfo, <-cha return machines, changes, nil } - -// CreateOrUpdateContainer creates a new container record or updates an existing one in the store database. -// The container is associated with the given machine ID that indicates which machine the container is running on. -func (s *Store) CreateOrUpdateContainer(ctx context.Context, c *container.Container, machineID string) error { - cJSON, err := json.Marshal(c) - if err != nil { - return fmt.Errorf("marshal container: %w", err) - } - - // Insert or update the container record if the container or machine ID has changed. - res, err := s.corro.ExecContext(ctx, ` - INSERT INTO containers (id, container, machine_id, sync_status, updated_at) - VALUES (?, ?, ?, ?, datetime('now')) - ON CONFLICT (id) DO UPDATE SET container = excluded.container, - machine_id = excluded.machine_id, - sync_status = excluded.sync_status, - updated_at = excluded.updated_at - WHERE containers.container != excluded.container - OR containers.machine_id != excluded.machine_id`, - c.ID, string(cJSON), machineID, SyncStatusSynced) - if err != nil { - return fmt.Errorf("upsert query: %w", err) - } - if res.RowsAffected > 0 { - slog.Debug("Container record updated in store DB.", "id", c.ID, "machine_id", machineID) - } - - return nil -}