Skip to content

Commit 93f0859

Browse files
ItsNotGoodNameItsNotGoodName
committed
refactor: config sync to db
1 parent b734d9c commit 93f0859

File tree

6 files changed

+110
-87
lines changed

6 files changed

+110
-87
lines changed

‎internal/jet/table/rules_to_endpoints.go

+2-8
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎internal/repo/internal.go

+101-60
Original file line numberDiff line numberDiff line change
@@ -20,30 +20,44 @@ func InternalSync(
2020
rules []models.Rule,
2121
ruleToEndpoints map[string][]string,
2222
) error {
23+
tx, err := db.BeginTx(ctx, nil)
24+
if err != nil {
25+
return err
26+
}
27+
defer tx.Rollback()
28+
2329
updatedAt := models.NewTime(time.Now())
2430

2531
for _, end := range endpoints {
26-
if err := internalEndpointUpsert(ctx, db, end, updatedAt); err != nil {
32+
if err := internalEndpointUpsert(ctx, tx, end, updatedAt); err != nil {
2733
return err
2834
}
2935
}
3036

3137
for _, rule := range rules {
32-
if err := internalRuleUpsert(ctx, db, rule, updatedAt); err != nil {
38+
if err := internalRuleUpsert(ctx, tx, rule, updatedAt); err != nil {
3339
return err
3440
}
3541
}
3642

43+
if err := internalDeleteOlderThan(ctx, tx, updatedAt); err != nil {
44+
return err
45+
}
46+
3747
for k, v := range ruleToEndpoints {
38-
if err := internalRuleEndpointsUpsert(ctx, db, k, v, updatedAt); err != nil {
48+
if err := internalRuleEndpointsInsert(ctx, tx, k, v); err != nil {
3949
return err
4050
}
4151
}
4252

43-
return internalDeleteOlderThan(ctx, db, updatedAt)
53+
if err := tx.Commit(); err != nil {
54+
return err
55+
}
56+
57+
return nil
4458
}
4559

46-
func internalEndpointUpsert(ctx context.Context, db database.Querier, r models.Endpoint, updatedAt models.Time) error {
60+
func internalEndpointUpsert(ctx context.Context, tx database.QuerierTx, r models.Endpoint, updatedAt models.Time) error {
4761
m := struct {
4862
models.Endpoint
4963
UpdatedAt models.Time
@@ -53,8 +67,9 @@ func internalEndpointUpsert(ctx context.Context, db database.Querier, r models.E
5367
UpdatedAt: updatedAt,
5468
CreatedAt: models.NewTime(time.Now()),
5569
}
56-
_, err := Endpoints.
57-
INSERT(
70+
71+
res, err := Endpoints.
72+
UPDATE(
5873
Endpoints.Internal,
5974
Endpoints.InternalID,
6075
Endpoints.Name,
@@ -65,26 +80,41 @@ func internalEndpointUpsert(ctx context.Context, db database.Querier, r models.E
6580
Endpoints.Kind,
6681
Endpoints.Config,
6782
Endpoints.UpdatedAt,
68-
Endpoints.CreatedAt,
6983
).
7084
MODEL(m).
71-
ON_CONFLICT(Endpoints.InternalID).
72-
DO_UPDATE(SET(
73-
Endpoints.InternalID.SET(Endpoints.EXCLUDED.InternalID),
74-
Endpoints.Name.SET(Endpoints.EXCLUDED.Name),
75-
Endpoints.AttachmentDisable.SET(Endpoints.EXCLUDED.AttachmentDisable),
76-
Endpoints.TextDisable.SET(Endpoints.EXCLUDED.TextDisable),
77-
Endpoints.TitleTemplate.SET(Endpoints.EXCLUDED.TitleTemplate),
78-
Endpoints.BodyTemplate.SET(Endpoints.EXCLUDED.BodyTemplate),
79-
Endpoints.Kind.SET(Endpoints.EXCLUDED.Kind),
80-
Endpoints.Config.SET(Endpoints.EXCLUDED.Config),
81-
Endpoints.UpdatedAt.SET(Endpoints.EXCLUDED.UpdatedAt),
82-
)).
83-
ExecContext(ctx, db)
84-
return err
85+
WHERE(Endpoints.InternalID.EQ(String(r.InternalID.String))).
86+
ExecContext(ctx, tx)
87+
if err != nil {
88+
return err
89+
}
90+
91+
count, err := res.RowsAffected()
92+
if count == 0 {
93+
_, err := Endpoints.
94+
INSERT(
95+
Endpoints.Internal,
96+
Endpoints.InternalID,
97+
Endpoints.Name,
98+
Endpoints.AttachmentDisable,
99+
Endpoints.TextDisable,
100+
Endpoints.TitleTemplate,
101+
Endpoints.BodyTemplate,
102+
Endpoints.Kind,
103+
Endpoints.Config,
104+
Endpoints.UpdatedAt,
105+
Endpoints.CreatedAt,
106+
).
107+
MODEL(m).
108+
ExecContext(ctx, tx)
109+
if err != nil {
110+
return err
111+
}
112+
}
113+
114+
return nil
85115
}
86116

87-
func internalRuleUpsert(ctx context.Context, db database.Querier, r models.Rule, updatedAt models.Time) error {
117+
func internalRuleUpsert(ctx context.Context, tx database.QuerierTx, r models.Rule, updatedAt models.Time) error {
88118
m := struct {
89119
models.Rule
90120
UpdatedAt models.Time
@@ -94,47 +124,68 @@ func internalRuleUpsert(ctx context.Context, db database.Querier, r models.Rule,
94124
UpdatedAt: updatedAt,
95125
CreatedAt: models.NewTime(time.Now()),
96126
}
97-
_, err := Rules.
98-
INSERT(
127+
128+
res, err := Rules.
129+
UPDATE(
99130
Rules.Internal,
100131
Rules.InternalID,
101132
Rules.Name,
102133
Rules.Expression,
103-
Rules.Enable,
104134
Rules.UpdatedAt,
105-
Rules.CreatedAt,
106135
).
107136
MODEL(m).
108-
ON_CONFLICT(Rules.InternalID).
109-
DO_UPDATE(SET(
110-
Rules.Name.SET(Rules.EXCLUDED.Name),
111-
Rules.Expression.SET(Rules.EXCLUDED.Expression),
112-
Rules.UpdatedAt.SET(Rules.EXCLUDED.UpdatedAt),
113-
)).
114-
ExecContext(ctx, db)
137+
WHERE(Rules.InternalID.EQ(String(r.InternalID.String))).
138+
ExecContext(ctx, tx)
139+
if err != nil {
140+
return err
141+
}
142+
143+
count, err := res.RowsAffected()
144+
if count == 0 {
145+
_, err := Rules.
146+
INSERT(
147+
Rules.Internal,
148+
Rules.InternalID,
149+
Rules.Name,
150+
Rules.Expression,
151+
Rules.Enable,
152+
Rules.UpdatedAt,
153+
Rules.CreatedAt,
154+
).
155+
MODEL(m).
156+
ExecContext(ctx, tx)
157+
if err != nil {
158+
return err
159+
}
160+
}
115161

116162
return err
117163
}
118164

119-
func internalRuleEndpointsUpsert(ctx context.Context, db database.Querier, ruleInternalID string, endpointInternalIDs []string, updatedAt models.Time) error {
120-
if len(endpointInternalIDs) == 0 {
121-
return nil
165+
func internalRuleEndpointsInsert(ctx context.Context, tx database.QuerierTx, ruleInternalID string, endpointInternalIDs []string) error {
166+
_, err := RulesToEndpoints.
167+
DELETE().
168+
WHERE(AND(
169+
RulesToEndpoints.RuleID.IN(Rules.SELECT(Rules.ID).WHERE(Rules.InternalID.EQ(String(ruleInternalID)))),
170+
RulesToEndpoints.Internal.EQ(Bool(true)),
171+
)).
172+
ExecContext(ctx, tx)
173+
if err != nil {
174+
return err
122175
}
123176

124177
for _, endpointInternalID := range endpointInternalIDs {
125-
// TODO: refactor this
126-
res, err := db.ExecContext(ctx, `
178+
res, err := tx.ExecContext(ctx, `
127179
INSERT INTO rules_to_endpoints (
128180
internal,
129181
rule_id,
130-
endpoint_id,
131-
updated_at,
132-
created_at
133-
) SELECT "1" AS internal, rules.id AS rule_id, endpoints.id AS endpoint_id, ?, ?
182+
endpoint_id
183+
) SELECT "1" AS internal, rules.id AS rule_id, endpoints.id AS endpoint_id
134184
FROM rules, endpoints
135-
WHERE rules.internal_id=? AND endpoints.internal_id IN (?)
136-
ON CONFLICT (rule_id, endpoint_id) DO UPDATE SET updated_at=EXCLUDED.updated_at, internal=EXCLUDED.internal
137-
`, updatedAt, updatedAt, ruleInternalID, endpointInternalID)
185+
WHERE rules.internal_id=? AND endpoints.internal_id IN (?)
186+
LIMIT 1
187+
ON CONFLICT (rule_id, endpoint_id) DO UPDATE SET internal=EXCLUDED.internal
188+
`, ruleInternalID, endpointInternalID)
138189
if err != nil {
139190
return err
140191
}
@@ -150,24 +201,14 @@ func internalRuleEndpointsUpsert(ctx context.Context, db database.Querier, ruleI
150201
return nil
151202
}
152203

153-
func internalDeleteOlderThan(ctx context.Context, db database.Querier, date models.Time) error {
154-
_, err := RulesToEndpoints.
155-
DELETE().
156-
WHERE(AND(
157-
RulesToEndpoints.Internal.IS_TRUE(),
158-
RulesToEndpoints.UpdatedAt.LT(RawTimestamp(muhTypeAffinity(date))),
159-
)).
160-
ExecContext(ctx, db)
161-
if err != nil {
162-
return err
163-
}
164-
_, err = Rules.
204+
func internalDeleteOlderThan(ctx context.Context, tx database.QuerierTx, date models.Time) error {
205+
_, err := Rules.
165206
DELETE().
166207
WHERE(AND(
167208
Rules.Internal.IS_TRUE(),
168209
Rules.UpdatedAt.LT(RawTimestamp(muhTypeAffinity(date))),
169210
)).
170-
ExecContext(ctx, db)
211+
ExecContext(ctx, tx)
171212
if err != nil {
172213
return err
173214
}
@@ -177,6 +218,6 @@ func internalDeleteOlderThan(ctx context.Context, db database.Querier, date mode
177218
Endpoints.Internal.IS_TRUE(),
178219
Endpoints.UpdatedAt.LT(RawTimestamp(muhTypeAffinity(date))),
179220
)).
180-
ExecContext(ctx, db)
221+
ExecContext(ctx, tx)
181222
return err
182223
}

‎internal/repo/rule.go

+4-8
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ func RuleCreate(ctx context.Context, db database.Querier, rule models.Rule, endp
5656
return 0, err
5757
}
5858

59-
err = ruleEndpointsSet(ctx, tx, now, ruleID, endpoints)
59+
err = ruleEndpointsSet(ctx, tx, ruleID, endpoints)
6060
if err != nil {
6161
return 0, err
6262
}
@@ -143,17 +143,15 @@ func RuleEndpointsSet(ctx context.Context, db database.Querier, ruleID int64, en
143143
return err
144144
}
145145

146-
now := models.NewTime(time.Now())
147-
148-
err = ruleEndpointsSet(ctx, tx, now, ruleID, endpointIDs)
146+
err = ruleEndpointsSet(ctx, tx, ruleID, endpointIDs)
149147
if err != nil {
150148
return err
151149
}
152150

153151
return tx.Commit()
154152
}
155153

156-
func ruleEndpointsSet(ctx context.Context, db database.QuerierTx, now models.Time, ruleID int64, endpointIDs []int64) error {
154+
func ruleEndpointsSet(ctx context.Context, db database.QuerierTx, ruleID int64, endpointIDs []int64) error {
157155
if len(endpointIDs) == 0 {
158156
return nil
159157
}
@@ -163,11 +161,9 @@ func ruleEndpointsSet(ctx context.Context, db database.QuerierTx, now models.Tim
163161
RulesToEndpoints.Internal,
164162
RulesToEndpoints.RuleID,
165163
RulesToEndpoints.EndpointID,
166-
RulesToEndpoints.UpdatedAt,
167-
RulesToEndpoints.CreatedAt,
168164
)
169165
for _, v := range endpointIDs {
170-
stmt = stmt.VALUES(false, ruleID, v, now, now)
166+
stmt = stmt.VALUES(false, ruleID, v)
171167
}
172168

173169
_, err := stmt.ExecContext(ctx, db)

‎migrations/20230912221604_initial.sql ‎migrations/20230913004557_initial.sql

+1-1
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ CREATE TABLE `rules` (`id` integer NOT NULL PRIMARY KEY AUTOINCREMENT, `internal
1212
-- create index "rules_internal_id_idx" to table: "rules"
1313
CREATE UNIQUE INDEX `rules_internal_id_idx` ON `rules` (`internal_id`);
1414
-- create "rules_to_endpoints" table
15-
CREATE TABLE `rules_to_endpoints` (`internal` boolean NOT NULL, `rule_id` integer NOT NULL, `endpoint_id` integer NOT NULL, `updated_at` datetime NOT NULL, `created_at` datetime NOT NULL, CONSTRAINT `endpoint_id` FOREIGN KEY (`endpoint_id`) REFERENCES `endpoints` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE, CONSTRAINT `rule_id` FOREIGN KEY (`rule_id`) REFERENCES `rules` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE);
15+
CREATE TABLE `rules_to_endpoints` (`internal` boolean NOT NULL, `rule_id` integer NOT NULL, `endpoint_id` integer NOT NULL, CONSTRAINT `endpoint_id` FOREIGN KEY (`endpoint_id`) REFERENCES `endpoints` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE, CONSTRAINT `rule_id` FOREIGN KEY (`rule_id`) REFERENCES `rules` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE);
1616
-- create index "rules_to_endpoints_rule_id_endpoint_id_idx" to table: "rules_to_endpoints"
1717
CREATE UNIQUE INDEX `rules_to_endpoints_rule_id_endpoint_id_idx` ON `rules_to_endpoints` (`rule_id`, `endpoint_id`);
1818
-- create "traces" table

‎migrations/atlas.sum

+2-2
Original file line numberDiff line numberDiff line change
@@ -1,2 +1,2 @@
1-
h1:WtkVjiwJteFNB0H2oF5A374xNdzoHn9MtUHwjE5C/rs=
2-
20230912221604_initial.sql h1:P0saNiMYasG9Ik82TWjpcP625oCI7wStswRjXfv/mCg=
1+
h1:RPpLniv2auRZdJ0jqZgi3EOZ4qrLs6YvyxGiXnpn4kA=
2+
20230913004557_initial.sql h1:zLMyXQpXiewvp3c7mjvsLTHyDlCui9UoNBfZcgwkxPI=

‎migrations/schema.hcl

-8
Original file line numberDiff line numberDiff line change
@@ -186,14 +186,6 @@ table "rules_to_endpoints" {
186186
null = false
187187
type = integer
188188
}
189-
column "updated_at" {
190-
null = false
191-
type = datetime
192-
}
193-
column "created_at" {
194-
null = false
195-
type = datetime
196-
}
197189
foreign_key "endpoint_id" {
198190
columns = [column.endpoint_id]
199191
ref_columns = [table.endpoints.column.id]

0 commit comments

Comments
 (0)