nathan/vctp2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

only check unprocessed events from last 1 day
Some checks are pending
CI / Test (push) Waiting to run
Details
CI / End-to-End (push) Waiting to run
Details
CI / Publish Docker (push) Blocked by required conditions
Details
CI / Lint (push) Waiting to run
Details
continuous-integration/drone/push Build is passing
Details

Browse Source
This commit is contained in:
Nathan Coad
2024-09-25 12:55:13 +10:00
parent 47bc8acace
commit c7c890f6bb
3 changed files with 8 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
db/queries/query.sql
View File

@@ -50,6 +50,7 @@ ORDER BY "EventTime";
-- name: ListUnprocessedEvents :many -- name: ListUnprocessedEvents :many
SELECT * FROM "Events" SELECT * FROM "Events"
WHERE "Processed" = 0 WHERE "Processed" = 0
AND "EventTime" > sqlc.arg('eventTime')
ORDER BY "EventTime"; ORDER BY "EventTime";
-- name: UpdateEventsProcessed :exec -- name: UpdateEventsProcessed :exec

5
db/queries/query.sql.go
View File

@@ -422,11 +422,12 @@ func (q *Queries) ListInventory(ctx context.Context) ([]Inventory, error) {
const listUnprocessedEvents = `-- name: ListUnprocessedEvents :many const listUnprocessedEvents = `-- name: ListUnprocessedEvents :many
SELECT Eid, CloudId, Source, EventTime, ChainId, VmId, EventKey, DatacenterName, ComputeResourceName, UserName, Processed, DatacenterId, ComputeResourceId, VmName, EventType FROM "Events" SELECT Eid, CloudId, Source, EventTime, ChainId, VmId, EventKey, DatacenterName, ComputeResourceName, UserName, Processed, DatacenterId, ComputeResourceId, VmName, EventType FROM "Events"
WHERE "Processed" = 0 WHERE "Processed" = 0
AND "EventTime" > ?1
ORDER BY "EventTime" ORDER BY "EventTime"
` `
func (q *Queries) ListUnprocessedEvents(ctx context.Context) ([]Events, error) { func (q *Queries) ListUnprocessedEvents(ctx context.Context, eventtime sql.NullInt64) ([]Events, error) {
rows, err := q.db.QueryContext(ctx, listUnprocessedEvents) rows, err := q.db.QueryContext(ctx, listUnprocessedEvents, eventtime)
if err != nil { if err != nil {
return nil, err return nil, err
} }

6
internal/tasks/processEvents.go
View File

@@ -31,10 +31,12 @@ func (c *CronTask) RunVmCheck(ctx context.Context, logger *slog.Logger) error {
folderPath string folderPath string
) )
logger.Debug("Started Events processing", "time", time.Now()) dateCmp := time.Now().AddDate(0, 0, -1).Unix()
logger.Debug("Started Events processing", "time", time.Now(), "since", dateCmp)
// Query events table // Query events table
events, err := c.Database.Queries().ListUnprocessedEvents(ctx) events, err := c.Database.Queries().ListUnprocessedEvents(ctx,
sql.NullInt64{Int64: dateCmp, Valid: dateCmp > 0})
if err != nil { if err != nil {
logger.Error("Unable to query for unprocessed events", "error", err) logger.Error("Unable to query for unprocessed events", "error", err)
return nil // TODO - what to do with this error? return nil // TODO - what to do with this error?