rohit/probo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor sql row can using collectable rows

Browse Source 
Signed-off-by: gearnode <bryan@frimin.fr>
This commit is contained in:
gearnode
2025-03-01 23:09:15 +01:00
parent 67edb3783e
commit bb463be895
13 changed files with 262 additions and 477 deletions
Download Patch File Download Diff File Expand all files Collapse all files

92
pkg/probo/coredata/control.go
View File

@@ -16,7 +16,6 @@ package coredata
import ( import (
"context" "context"
"errors"
"fmt" "fmt"
"maps" "maps"
"time" "time"
@@ -30,16 +29,16 @@ import (
type ( type (
Control struct { Control struct {
ID gid.GID ID gid.GID `db:"id"`
FrameworkID gid.GID FrameworkID gid.GID `db:"framework_id"`
Category string Category string `db:"category"`
Name string Name string `db:"name"`
Description string Description string `db:"description"`
State ControlState State ControlState `db:"state"`
ContentRef string ContentRef string `db:"content_ref"`
CreatedAt time.Time CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time UpdatedAt time.Time `db:"updated_at"`
Version int Version int `db:"version"`
} }
Controls []*Control Controls []*Control
@@ -57,22 +56,7 @@ func (c Control) CursorKey() page.CursorKey {
return page.NewCursorKey(c.ID, c.CreatedAt) return page.NewCursorKey(c.ID, c.CreatedAt)
} }
func (c *Control) scan(r pgx.Row) error { func (c *Control) LoadByID(
return r.Scan(
&c.ID,
&c.FrameworkID,
&c.Category,
&c.Name,
&c.Description,
&c.State,
&c.ContentRef,
&c.CreatedAt,
&c.UpdatedAt,
&c.Version,
)
}
func (v *Control) LoadByID(
ctx context.Context, ctx context.Context,
conn pg.Conn, conn pg.Conn,
scope *Scope, scope *Scope,
@@ -119,14 +103,17 @@ LIMIT 1;
args := pgx.NamedArgs{"control_id": controlID} args := pgx.NamedArgs{"control_id": controlID}
maps.Copy(args, scope.SQLArguments()) maps.Copy(args, scope.SQLArguments())
r := conn.QueryRow(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil {
c2 := Control{} return fmt.Errorf("cannot query controls: %w", err)
if err := c2.scan(r); err != nil {
return err
} }
*v = c2 control, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[Control])
if err != nil {
return fmt.Errorf("cannot collect controls: %w", err)
}
*c = control
return nil return nil
} }
@@ -222,24 +209,14 @@ WHERE
maps.Copy(args, scope.SQLArguments()) maps.Copy(args, scope.SQLArguments())
maps.Copy(args, cursor.SQLArguments()) maps.Copy(args, cursor.SQLArguments())
r, err := conn.Query(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil { if err != nil {
return err return fmt.Errorf("cannot query controls: %w", err)
}
defer r.Close()
controls := Controls{}
for r.Next() {
control := &Control{}
if err := control.scan(r); err != nil {
return err
}
controls = append(controls, control)
} }
if err := r.Err(); err != nil { controls, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[Control])
return err if err != nil {
return fmt.Errorf("cannot collect controls: %w", err)
} }
*c = controls *c = controls
@@ -306,16 +283,17 @@ RETURNING
maps.Copy(args, scope.SQLArguments()) maps.Copy(args, scope.SQLArguments())
r := conn.QueryRow(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil {
c2 := Control{} return fmt.Errorf("cannot query controls: %w", err)
if err := c2.scan(r); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return ErrConcurrentModification
}
return err
} }
*c = c2 control, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[Control])
if err != nil {
return fmt.Errorf("cannot collect controls: %w", err)
}
*c = control
return nil return nil
} }

36
pkg/probo/coredata/control_state_transition.go
View File

@@ -29,7 +29,7 @@ type (
ControlStateTransition struct { ControlStateTransition struct {
StateTransition[ControlState] StateTransition[ControlState]
ControlID gid.GID ControlID gid.GID `db:"control_id"`
} }
ControlStateTransitions []*ControlStateTransition ControlStateTransitions []*ControlStateTransition
@@ -39,18 +39,6 @@ func (cst ControlStateTransition) CursorKey() page.CursorKey {
return page.NewCursorKey(cst.ID, cst.CreatedAt) return page.NewCursorKey(cst.ID, cst.CreatedAt)
} }
func (cst *ControlStateTransition) scan(r pgx.Row) error {
return r.Scan(
&cst.ID,
&cst.ControlID,
&cst.FromState,
&cst.ToState,
&cst.Reason,
&cst.CreatedAt,
&cst.UpdatedAt,
)
}
func (cst ControlStateTransition) Insert( func (cst ControlStateTransition) Insert(
ctx context.Context, ctx context.Context,
conn pg.Conn, conn pg.Conn,
@@ -119,27 +107,17 @@ WHERE
args := pgx.NamedArgs{"control_id": controlID} args := pgx.NamedArgs{"control_id": controlID}
maps.Copy(args, scope.SQLArguments()) maps.Copy(args, scope.SQLArguments())
r, err := conn.Query(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil { if err != nil {
return err return fmt.Errorf("cannot query control state transitions: %w", err)
}
defer r.Close()
controlStateTransitions := ControlStateTransitions{}
for r.Next() {
controlStateTransition := &ControlStateTransition{}
if err := controlStateTransition.scan(r); err != nil {
return err
}
controlStateTransitions = append(controlStateTransitions, controlStateTransition)
} }
if err := r.Err(); err != nil { controlStateTransitions, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[ControlStateTransition])
return err if err != nil {
return fmt.Errorf("cannot collect control state transitions: %w", err)
} }
*cst = controlStateTransitions *cst = ControlStateTransitions(controlStateTransitions)
return nil return nil
} }

67
pkg/probo/coredata/evidence.go
View File

@@ -28,15 +28,15 @@ import (
type ( type (
Evidence struct { Evidence struct {
ID gid.GID ID gid.GID `db:"id"`
TaskID gid.GID TaskID gid.GID `db:"task_id"`
State EvidenceState State EvidenceState `db:"state"`
ObjectKey string ObjectKey string `db:"object_key"`
MimeType string MimeType string `db:"mime_type"`
Size uint64 Size uint64 `db:"size"`
Filename string Filename string `db:"filename"`
CreatedAt time.Time CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time UpdatedAt time.Time `db:"updated_at"`
} }
Evidences []*Evidence Evidences []*Evidence
@@ -46,20 +46,6 @@ func (e Evidence) CursorKey() page.CursorKey {
return page.NewCursorKey(e.ID, e.CreatedAt) return page.NewCursorKey(e.ID, e.CreatedAt)
} }
func (e *Evidence) scan(r pgx.Row) error {
return r.Scan(
&e.ID,
&e.TaskID,
&e.State,
&e.ObjectKey,
&e.MimeType,
&e.Size,
&e.Filename,
&e.CreatedAt,
&e.UpdatedAt,
)
}
func (e Evidence) Insert( func (e Evidence) Insert(
ctx context.Context, ctx context.Context,
conn pg.Conn, conn pg.Conn,
@@ -149,14 +135,17 @@ LIMIT 1;
args := pgx.NamedArgs{"evidence_id": evidenceID} args := pgx.NamedArgs{"evidence_id": evidenceID}
maps.Copy(args, scope.SQLArguments()) maps.Copy(args, scope.SQLArguments())
r := conn.QueryRow(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil {
e2 := Evidence{} return fmt.Errorf("cannot query evidence: %w", err)
if err := e2.scan(r); err != nil {
return err
} }
*e = e2 evidence, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[Evidence])
if err != nil {
return fmt.Errorf("cannot collect evidence: %w", err)
}
*e = evidence
return nil return nil
} }
@@ -208,24 +197,14 @@ WHERE
maps.Copy(args, scope.SQLArguments()) maps.Copy(args, scope.SQLArguments())
maps.Copy(args, cursor.SQLArguments()) maps.Copy(args, cursor.SQLArguments())
r, err := conn.Query(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil { if err != nil {
return err return fmt.Errorf("cannot query evidence: %w", err)
}
defer r.Close()
evidences := Evidences{}
for r.Next() {
evidence := &Evidence{}
if err := evidence.scan(r); err != nil {
return err
}
evidences = append(evidences, evidence)
} }
if err := r.Err(); err != nil { evidences, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[Evidence])
return err if err != nil {
return fmt.Errorf("cannot collect evidence: %w", err)
} }
*e = evidences *e = evidences

34
pkg/probo/coredata/evidence_state_transition.go
View File

@@ -29,7 +29,7 @@ type (
EvidenceStateTransition struct { EvidenceStateTransition struct {
StateTransition[EvidenceState] StateTransition[EvidenceState]
EvidenceID gid.GID EvidenceID gid.GID `db:"evidence_id"`
} }
EvidenceStateTransitions []*EvidenceStateTransition EvidenceStateTransitions []*EvidenceStateTransition
@@ -39,18 +39,6 @@ func (cst EvidenceStateTransition) CursorKey() page.CursorKey {
return page.NewCursorKey(cst.ID, cst.CreatedAt) return page.NewCursorKey(cst.ID, cst.CreatedAt)
} }
func (cst *EvidenceStateTransition) scan(r pgx.Row) error {
return r.Scan(
&cst.ID,
&cst.EvidenceID,
&cst.FromState,
&cst.ToState,
&cst.Reason,
&cst.CreatedAt,
&cst.UpdatedAt,
)
}
func (est EvidenceStateTransition) Insert( func (est EvidenceStateTransition) Insert(
ctx context.Context, ctx context.Context,
conn pg.Conn, conn pg.Conn,
@@ -119,24 +107,14 @@ WHERE
args := pgx.NamedArgs{"evidence_id": evidenceID} args := pgx.NamedArgs{"evidence_id": evidenceID}
maps.Copy(args, scope.SQLArguments()) maps.Copy(args, scope.SQLArguments())
r, err := conn.Query(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil { if err != nil {
return err return fmt.Errorf("cannot query evidence state transitions: %w", err)
}
defer r.Close()
evidenceStateTransitions := EvidenceStateTransitions{}
for r.Next() {
evidenceStateTransition := &EvidenceStateTransition{}
if err := evidenceStateTransition.scan(r); err != nil {
return err
}
evidenceStateTransitions = append(evidenceStateTransitions, evidenceStateTransition)
} }
if err := r.Err(); err != nil { evidenceStateTransitions, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[EvidenceStateTransition])
return err if err != nil {
return fmt.Errorf("cannot collect evidence state transitions: %w", err)
} }
*cst = evidenceStateTransitions *cst = evidenceStateTransitions

84
pkg/probo/coredata/framework.go
View File

@@ -16,7 +16,6 @@ package coredata
import ( import (
"context" "context"
"errors"
"fmt" "fmt"
"maps" "maps"
"time" "time"
@@ -29,14 +28,14 @@ import (
type ( type (
Framework struct { Framework struct {
ID gid.GID ID gid.GID `db:"id"`
OrganizationID gid.GID OrganizationID gid.GID `db:"organization_id"`
Name string Name string `db:"name"`
Description string Description string `db:"description"`
ContentRef string ContentRef string `db:"content_ref"`
CreatedAt time.Time CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time UpdatedAt time.Time `db:"updated_at"`
Version int Version int `db:"version"`
} }
Frameworks []*Framework Frameworks []*Framework
@@ -52,19 +51,6 @@ func (f Framework) CursorKey() page.CursorKey {
return page.NewCursorKey(f.ID, f.CreatedAt) return page.NewCursorKey(f.ID, f.CreatedAt)
} }
func (f *Framework) scan(r pgx.Row) error {
return r.Scan(
&f.ID,
&f.OrganizationID,
&f.Name,
&f.Description,
&f.ContentRef,
&f.CreatedAt,
&f.UpdatedAt,
&f.Version,
)
}
func (f *Frameworks) LoadByOrganizationID( func (f *Frameworks) LoadByOrganizationID(
ctx context.Context, ctx context.Context,
conn pg.Conn, conn pg.Conn,
@@ -96,24 +82,14 @@ WHERE
maps.Copy(args, cursor.SQLArguments()) maps.Copy(args, cursor.SQLArguments())
maps.Copy(args, cursor.SQLArguments()) maps.Copy(args, cursor.SQLArguments())
r, err := conn.Query(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil { if err != nil {
return err return fmt.Errorf("cannot query frameworks: %w", err)
}
defer r.Close()
frameworks := Frameworks{}
for r.Next() {
framework := &Framework{}
if err := framework.scan(r); err != nil {
return err
}
frameworks = append(frameworks, framework)
} }
if err := r.Err(); err != nil { frameworks, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[Framework])
return err if err != nil {
return fmt.Errorf("cannot collect frameworks: %w", err)
} }
*f = frameworks *f = frameworks
@@ -149,14 +125,17 @@ LIMIT 1;
args := pgx.NamedArgs{"framework_id": frameworkID} args := pgx.NamedArgs{"framework_id": frameworkID}
maps.Copy(args, scope.SQLArguments()) maps.Copy(args, scope.SQLArguments())
r := conn.QueryRow(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil {
f2 := Framework{} return fmt.Errorf("cannot query frameworks: %w", err)
if err := f2.scan(r); err != nil {
return err
} }
*f = f2 framework, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[Framework])
if err != nil {
return fmt.Errorf("cannot collect framework: %w", err)
}
*f = framework
return nil return nil
} }
@@ -267,16 +246,17 @@ RETURNING
maps.Copy(args, scope.SQLArguments()) maps.Copy(args, scope.SQLArguments())
r := conn.QueryRow(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil {
f2 := Framework{} return fmt.Errorf("cannot query frameworks: %w", err)
if err := f2.scan(r); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return ErrConcurrentModification
}
return err
} }
*f = f2 framework, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[Framework])
if err != nil {
return fmt.Errorf("cannot collect framework: %w", err)
}
*f = framework
return nil return nil
} }

35
pkg/probo/coredata/organization.go
View File

@@ -27,24 +27,14 @@ import (
type ( type (
Organization struct { Organization struct {
ID gid.GID ID gid.GID `db:"id"`
Name string Name string `db:"name"`
LogoURL string LogoURL string `db:"logo_url"`
CreatedAt time.Time CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time UpdatedAt time.Time `db:"updated_at"`
} }
) )
func (o *Organization) scan(r pgx.Row) error {
return r.Scan(
&o.ID,
&o.Name,
&o.LogoURL,
&o.CreatedAt,
&o.UpdatedAt,
)
}
func (o *Organization) LoadByID( func (o *Organization) LoadByID(
ctx context.Context, ctx context.Context,
conn pg.Conn, conn pg.Conn,
@@ -71,14 +61,17 @@ LIMIT 1;
args := pgx.NamedArgs{"organization_id": organizationID} args := pgx.NamedArgs{"organization_id": organizationID}
maps.Copy(args, scope.SQLArguments()) maps.Copy(args, scope.SQLArguments())
r := conn.QueryRow(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil {
o2 := Organization{} return fmt.Errorf("cannot query organizations: %w", err)
if err := o2.scan(r); err != nil {
return err
} }
*o = o2 organization, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[Organization])
if err != nil {
return fmt.Errorf("cannot collect organization: %w", err)
}
*o = organization
return nil return nil
} }

87
pkg/probo/coredata/people.go
View File

@@ -16,7 +16,6 @@ package coredata
import ( import (
"context" "context"
"errors"
"fmt" "fmt"
"maps" "maps"
"time" "time"
@@ -29,15 +28,15 @@ import (
type ( type (
People struct { People struct {
ID gid.GID ID gid.GID `db:"id"`
OrganizationID gid.GID OrganizationID gid.GID `db:"organization_id"`
Kind PeopleKind Kind PeopleKind `db:"kind"`
FullName string FullName string `db:"full_name"`
PrimaryEmailAddress string PrimaryEmailAddress string `db:"primary_email_address"`
AdditionalEmailAddresses []string AdditionalEmailAddresses []string `db:"additional_email_addresses"`
CreatedAt time.Time CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time UpdatedAt time.Time `db:"updated_at"`
Version int Version int `db:"version"`
} }
Peoples []*People Peoples []*People
@@ -55,20 +54,6 @@ func (p People) CursorKey() page.CursorKey {
return page.NewCursorKey(p.ID, p.CreatedAt) return page.NewCursorKey(p.ID, p.CreatedAt)
} }
func (p *People) scan(r pgx.Row) error {
return r.Scan(
&p.ID,
&p.OrganizationID,
&p.Kind,
&p.FullName,
&p.PrimaryEmailAddress,
&p.AdditionalEmailAddresses,
&p.CreatedAt,
&p.UpdatedAt,
&p.Version,
)
}
func (p *People) LoadByID( func (p *People) LoadByID(
ctx context.Context, ctx context.Context,
conn pg.Conn, conn pg.Conn,
@@ -99,14 +84,17 @@ LIMIT 1;
args := pgx.NamedArgs{"people_id": peopleID} args := pgx.NamedArgs{"people_id": peopleID}
maps.Copy(args, scope.SQLArguments()) maps.Copy(args, scope.SQLArguments())
r := conn.QueryRow(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil {
p2 := People{} return fmt.Errorf("cannot query people: %w", err)
if err := p2.scan(r); err != nil {
return err
} }
*p = p2 people, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[People])
if err != nil {
return fmt.Errorf("cannot collect people: %w", err)
}
*p = people
return nil return nil
} }
@@ -206,24 +194,14 @@ WHERE
maps.Copy(args, cursor.SQLArguments()) maps.Copy(args, cursor.SQLArguments())
maps.Copy(args, cursor.SQLArguments()) maps.Copy(args, cursor.SQLArguments())
r, err := conn.Query(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil { if err != nil {
return err return fmt.Errorf("cannot query people: %w", err)
}
defer r.Close()
peoples := Peoples{}
for r.Next() {
people := &People{}
if err := people.scan(r); err != nil {
return err
}
peoples = append(peoples, people)
} }
if err := r.Err(); err != nil { peoples, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[People])
return err if err != nil {
return fmt.Errorf("cannot collect people: %w", err)
} }
*p = peoples *p = peoples
@@ -282,16 +260,17 @@ RETURNING
maps.Copy(args, scope.SQLArguments()) maps.Copy(args, scope.SQLArguments())
r := conn.QueryRow(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil {
p2 := People{} return fmt.Errorf("cannot query people: %w", err)
if err := p2.scan(r); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return ErrConcurrentModification
}
return err
} }
*p = p2 people, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[People])
if err != nil {
return fmt.Errorf("cannot collect people: %w", err)
}
*p = people
return nil return nil
} }

12
pkg/probo/coredata/state_transition.go
View File

@@ -21,11 +21,11 @@ import (
type ( type (
StateTransition[T any] struct { StateTransition[T any] struct {
ID gid.GID ID gid.GID `db:"id"`
FromState *T ToState T `db:"to_state"`
ToState T FromState *T `db:"from_state"`
Reason *string Reason *string `db:"reason"`
CreatedAt time.Time CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time UpdatedAt time.Time `db:"updated_at"`
} }
) )

65
pkg/probo/coredata/task.go
View File

@@ -29,14 +29,14 @@ import (
type ( type (
Task struct { Task struct {
ID gid.GID ID gid.GID `db:"id"`
ControlID gid.GID ControlID gid.GID `db:"control_id"`
Name string Name string `db:"name"`
Description string Description string `db:"description"`
State TaskState State TaskState `db:"state"`
ContentRef string ContentRef string `db:"content_ref"`
CreatedAt time.Time CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time UpdatedAt time.Time `db:"updated_at"`
} }
Tasks []*Task Tasks []*Task
@@ -46,19 +46,6 @@ func (t Task) CursorKey() page.CursorKey {
return page.NewCursorKey(t.ID, t.CreatedAt) return page.NewCursorKey(t.ID, t.CreatedAt)
} }
func (t *Task) scan(r pgx.Row) error {
return r.Scan(
&t.ID,
&t.ControlID,
&t.Name,
&t.Description,
&t.State,
&t.ContentRef,
&t.CreatedAt,
&t.UpdatedAt,
)
}
func (t *Task) LoadByID( func (t *Task) LoadByID(
ctx context.Context, ctx context.Context,
conn pg.Conn, conn pg.Conn,
@@ -122,14 +109,17 @@ LIMIT 1;
args := pgx.NamedArgs{"task_id": taskID} args := pgx.NamedArgs{"task_id": taskID}
maps.Copy(args, scope.SQLArguments()) maps.Copy(args, scope.SQLArguments())
r := conn.QueryRow(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil {
t2 := Task{} return fmt.Errorf("cannot query task: %w", err)
if err := t2.scan(r); err != nil {
return err
} }
*t = t2 task, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[Task])
if err != nil {
return fmt.Errorf("cannot collect task: %w", err)
}
*t = task
return nil return nil
} }
@@ -245,24 +235,14 @@ WHERE
maps.Copy(args, scope.SQLArguments()) maps.Copy(args, scope.SQLArguments())
maps.Copy(args, cursor.SQLArguments()) maps.Copy(args, cursor.SQLArguments())
r, err := conn.Query(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil { if err != nil {
return err return fmt.Errorf("cannot query tasks: %w", err)
}
defer r.Close()
tasks := Tasks{}
for r.Next() {
task := &Task{}
if err := task.scan(r); err != nil {
return err
}
tasks = append(tasks, task)
} }
if err := r.Err(); err != nil { tasks, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[Task])
return err if err != nil {
return fmt.Errorf("cannot collect tasks: %w", err)
} }
*t = tasks *t = tasks
@@ -275,7 +255,6 @@ func (t *Task) Delete(
conn pg.Conn, conn pg.Conn,
scope *Scope, scope *Scope,
) error { ) error {
// Use a single transaction with conditional logic to handle both cases
q := ` q := `
WITH control_count AS ( WITH control_count AS (
SELECT COUNT(*) AS count FROM controls_tasks WHERE task_id = @task_id SELECT COUNT(*) AS count FROM controls_tasks WHERE task_id = @task_id

35
pkg/probo/coredata/task_state_transition.go
View File

@@ -27,9 +27,8 @@ import (
type ( type (
TaskStateTransition struct { TaskStateTransition struct {
TaskID gid.GID
StateTransition[TaskState] StateTransition[TaskState]
TaskID gid.GID `db:"task_id"`
} }
TaskStateTransitions []*TaskStateTransition TaskStateTransitions []*TaskStateTransition
@@ -39,18 +38,6 @@ func (tst TaskStateTransition) CursorKey() page.CursorKey {
return page.NewCursorKey(tst.ID, tst.CreatedAt) return page.NewCursorKey(tst.ID, tst.CreatedAt)
} }
func (tst *TaskStateTransition) scan(r pgx.Row) error {
return r.Scan(
&tst.ID,
&tst.TaskID,
&tst.FromState,
&tst.ToState,
&tst.Reason,
&tst.CreatedAt,
&tst.UpdatedAt,
)
}
func (tst TaskStateTransition) Insert( func (tst TaskStateTransition) Insert(
ctx context.Context, ctx context.Context,
conn pg.Conn, conn pg.Conn,
@@ -119,24 +106,14 @@ WHERE
args := pgx.NamedArgs{"task_id": taskID} args := pgx.NamedArgs{"task_id": taskID}
maps.Copy(args, scope.SQLArguments()) maps.Copy(args, scope.SQLArguments())
r, err := conn.Query(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil { if err != nil {
return err return fmt.Errorf("cannot query task state transitions: %w", err)
}
defer r.Close()
taskStateTransitions := TaskStateTransitions{}
for r.Next() {
taskStateTransition := &TaskStateTransition{}
if err := taskStateTransition.scan(r); err != nil {
return err
}
taskStateTransitions = append(taskStateTransitions, taskStateTransition)
} }
if err := r.Err(); err != nil { taskStateTransitions, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[TaskStateTransition])
return err if err != nil {
return fmt.Errorf("cannot collect task state transitions: %w", err)
} }
*tst = taskStateTransitions *tst = taskStateTransitions

100
pkg/probo/coredata/vendor.go
View File

@@ -31,20 +31,20 @@ var ErrConcurrentModification = errors.New("concurrent modification")
type ( type (
Vendor struct { Vendor struct {
ID gid.GID ID gid.GID `db:"id"`
OrganizationID gid.GID OrganizationID gid.GID `db:"organization_id"`
Name string Name string `db:"name"`
Description string Description string `db:"description"`
ServiceStartAt time.Time ServiceStartAt time.Time `db:"service_start_at"`
ServiceTerminationAt *time.Time ServiceTerminationAt *time.Time `db:"service_termination_at"`
ServiceCriticality ServiceCriticality ServiceCriticality ServiceCriticality `db:"service_criticality"`
RiskTier RiskTier RiskTier RiskTier `db:"risk_tier"`
StatusPageURL *string StatusPageURL *string `db:"status_page_url"`
TermsOfServiceURL *string TermsOfServiceURL *string `db:"terms_of_service_url"`
PrivacyPolicyURL *string PrivacyPolicyURL *string `db:"privacy_policy_url"`
CreatedAt time.Time CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time UpdatedAt time.Time `db:"updated_at"`
Version int Version int `db:"version"`
} }
Vendors []*Vendor Vendors []*Vendor
@@ -67,25 +67,6 @@ func (v Vendor) CursorKey() page.CursorKey {
return page.NewCursorKey(v.ID, v.CreatedAt) return page.NewCursorKey(v.ID, v.CreatedAt)
} }
func (v *Vendor) scan(r pgx.Row) error {
return r.Scan(
&v.ID,
&v.OrganizationID,
&v.Name,
&v.Description,
&v.ServiceStartAt,
&v.ServiceTerminationAt,
&v.ServiceCriticality,
&v.RiskTier,
&v.StatusPageURL,
&v.TermsOfServiceURL,
&v.PrivacyPolicyURL,
&v.CreatedAt,
&v.UpdatedAt,
&v.Version,
)
}
func (v *Vendor) LoadByID( func (v *Vendor) LoadByID(
ctx context.Context, ctx context.Context,
conn pg.Conn, conn pg.Conn,
@@ -121,14 +102,18 @@ LIMIT 1;
args := pgx.NamedArgs{"vendor_id": vendorID} args := pgx.NamedArgs{"vendor_id": vendorID}
maps.Copy(args, scope.SQLArguments()) maps.Copy(args, scope.SQLArguments())
r := conn.QueryRow(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil {
return fmt.Errorf("cannot query vendor: %w", err)
}
defer rows.Close()
v2 := Vendor{} vendor, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[Vendor])
if err := v2.scan(r); err != nil { if err != nil {
return err return fmt.Errorf("cannot collect vendor: %w", err)
} }
*v = v2 *v = vendor
return nil return nil
} }
@@ -247,24 +232,14 @@ WHERE
maps.Copy(args, cursor.SQLArguments()) maps.Copy(args, cursor.SQLArguments())
maps.Copy(args, scope.SQLArguments()) maps.Copy(args, scope.SQLArguments())
r, err := conn.Query(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil { if err != nil {
return err return fmt.Errorf("cannot query vendors: %w", err)
}
defer r.Close()
vendors := Vendors{}
for r.Next() {
vendor := &Vendor{}
if err := vendor.scan(r); err != nil {
return err
}
vendors = append(vendors, vendor)
} }
if err := r.Err(); err != nil { vendors, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[Vendor])
return err if err != nil {
return fmt.Errorf("cannot collect vendors: %w", err)
} }
*v = vendors *v = vendors
@@ -348,16 +323,17 @@ RETURNING
maps.Copy(args, scope.SQLArguments()) maps.Copy(args, scope.SQLArguments())
r := conn.QueryRow(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil {
v2 := Vendor{} return fmt.Errorf("cannot query vendor: %w", err)
if err := v2.scan(r); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return ErrConcurrentModification
}
return err
} }
*v = v2 vendor, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[Vendor])
if err != nil {
return fmt.Errorf("cannot collect vendor: %w", err)
}
*v = vendor
return nil return nil
} }

35
pkg/usrmgr/coredata/session.go
View File

@@ -16,6 +16,7 @@ package coredata
import ( import (
"context" "context"
"fmt"
"time" "time"
"github.com/getprobo/probo/pkg/gid" "github.com/getprobo/probo/pkg/gid"
@@ -26,11 +27,11 @@ import (
type ( type (
Session struct { Session struct {
ID gid.GID ID gid.GID `db:"id"`
UserID gid.GID UserID gid.GID `db:"user_id"`
ExpiredAt time.Time ExpiredAt time.Time `db:"expired_at"`
CreatedAt time.Time CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time UpdatedAt time.Time `db:"updated_at"`
} }
) )
@@ -38,16 +39,6 @@ func (s Session) CursorKey() page.CursorKey {
return page.NewCursorKey(s.ID, s.CreatedAt) return page.NewCursorKey(s.ID, s.CreatedAt)
} }
func (s *Session) scan(r pgx.Row) error {
return r.Scan(
&s.ID,
&s.UserID,
&s.ExpiredAt,
&s.CreatedAt,
&s.UpdatedAt,
)
}
func (s *Session) LoadByID( func (s *Session) LoadByID(
ctx context.Context, ctx context.Context,
conn pg.Conn, conn pg.Conn,
@@ -69,14 +60,16 @@ LIMIT 1;
args := pgx.NamedArgs{"session_id": sessionID} args := pgx.NamedArgs{"session_id": sessionID}
r := conn.QueryRow(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil {
s2 := Session{} return fmt.Errorf("cannot query session: %w", err)
if err := s2.scan(r); err != nil {
return err
} }
*s = s2 session, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[Session])
if err != nil {
return fmt.Errorf("cannot collect session: %w", err)
}
*s = session
return nil return nil
} }

57
pkg/usrmgr/coredata/user.go
View File

@@ -16,6 +16,7 @@ package coredata
import ( import (
"context" "context"
"fmt"
"time" "time"
"github.com/getprobo/probo/pkg/gid" "github.com/getprobo/probo/pkg/gid"
@@ -26,13 +27,13 @@ import (
type ( type (
User struct { User struct {
ID gid.GID ID gid.GID `db:"id"`
EmailAddress string EmailAddress string `db:"email_address"`
HashedPassword []byte HashedPassword []byte `db:"hashed_password"`
FullName string FullName string `db:"fullname"`
OrganizationID gid.GID OrganizationID gid.GID `db:"organization_id"`
CreatedAt time.Time CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time UpdatedAt time.Time `db:"updated_at"`
} }
) )
@@ -40,18 +41,6 @@ func (u User) CursorKey() page.CursorKey {
return page.NewCursorKey(u.ID, u.CreatedAt) return page.NewCursorKey(u.ID, u.CreatedAt)
} }
func (u *User) scan(r pgx.Row) error {
return r.Scan(
&u.ID,
&u.EmailAddress,
&u.HashedPassword,
&u.FullName,
&u.OrganizationID,
&u.CreatedAt,
&u.UpdatedAt,
)
}
func (u *User) LoadByEmail( func (u *User) LoadByEmail(
ctx context.Context, ctx context.Context,
conn pg.Conn, conn pg.Conn,
@@ -75,14 +64,17 @@ LIMIT 1;
args := pgx.NamedArgs{"user_email": email} args := pgx.NamedArgs{"user_email": email}
r := conn.QueryRow(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil {
u2 := User{} return fmt.Errorf("cannot query user: %w", err)
if err := u2.scan(r); err != nil {
return err
} }
*u = u2 user, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[User])
if err != nil {
return fmt.Errorf("cannot collect user: %w", err)
}
*u = user
return nil return nil
} }
@@ -110,14 +102,17 @@ LIMIT 1;
args := pgx.NamedArgs{"user_id": userID} args := pgx.NamedArgs{"user_id": userID}
r := conn.QueryRow(ctx, q, args) rows, err := conn.Query(ctx, q, args)
if err != nil {
u2 := User{} return fmt.Errorf("cannot query user: %w", err)
if err := u2.scan(r); err != nil {
return err
} }
*u = u2 user, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[User])
if err != nil {
return fmt.Errorf("cannot collect user: %w", err)
}
*u = user
return nil return nil
} }