Integrated, working.

This integrates the liteq, and it prevents duplicates in a way that
matches my use-case.

I might try and push things back out to a separate module, but for now,
this will do.

.gitignore

@@ -0,0 +1,2 @@
+*.db*
+*.sqlite*

Makefile

@@ -4,3 +4,4 @@ test: generate
 generate:
 	cd internal/domain64 ; make generate
 	cd internal/liteq ; sqlc generate
+	cd internal/liteq ; make schema-const

cmd/api/main.go

@@ -12,9 +12,11 @@ import (
 
 	_ "modernc.org/sqlite"
 
-	liteq "git.jadud.com/grosbeak/internal/liteq"
+	liteq "git.jadud.com/jadudm/grosbeak/internal/liteq"
 )
 
+type queueWorker func(ctx context.Context, job *liteq.Job) error
+
 func Fetch(ctx context.Context, job *liteq.Job) error {
 	n := rand.Intn(50)
 
@@ -23,38 +25,48 @@ func Fetch(ctx context.Context, job *liteq.Job) error {
 	return nil
 }
 
-func runWorkers(queue *liteq.JobQueue) {
+func runQ(queue *liteq.JobQueue, queueName string, worker queueWorker) {
-	go queue.Consume(context.Background(), liteq.ConsumeParams{
+	for {
-		Queue:             "fetch",
+		err := queue.Consume(context.Background(), liteq.ConsumeParams{
-		PoolSize:          3,
+			Queue:             queueName,
-		VisibilityTimeout: 200000,
+			PoolSize:          3,
-		Worker:            Fetch,
+			VisibilityTimeout: 20,
-	})
+			Worker:            worker,
+		})
+		if err != nil {
+			log.Printf("runQ/%s: %w", queueName, err.Error())
+			time.Sleep(2 * time.Second)
+		}
+		time.Sleep(1 * time.Second)
+	}
 }
 
 func Entre(queue *liteq.JobQueue, chUrl <-chan string) {
 	ctx := context.Background()
 	for {
 		url := <-chUrl
-		log.Println("Entre", url)
-		// Don't duplicate jobs on the same day of the year.
 		n := time.Now()
-		queue.QueueJob(ctx, liteq.QueueJobParams{
+		ignore_tag := fmt.Sprintf("%s:%d:%d", url, n.Year(), n.YearDay())
+		log.Println("entre", url, ignore_tag)
+		// Don't duplicate jobs on the same day of the year.
+		err := queue.QueueJob(ctx, liteq.QueueJobParams{
 			Queue: "fetch",
 			// This only works for things in the `queued` state
-			DedupingKey: liteq.IgnoreDuplicate(fmt.Sprintf("%s:%d:%d", url, n.Year(), n.YearDay())),
+			DedupingKey: liteq.IgnoreDuplicate(ignore_tag),
 			Job:         url,
 		})
+		if err != nil {
+			log.Println("entre err", err.Error())
+		}
 	}
 }
 
-func main() {
+func setupLiteQ() *liteq.JobQueue {
-	// Don't let `main()` exit
-	wg := &sync.WaitGroup{}
-	wg.Add(1)
 
 	// FIXME: This path needs to come from the env.
 	liteqDB, err := sql.Open("sqlite", "liteq.db")
+	liteqDB.SetMaxOpenConns(1)
+
 	if err != nil {
 		fmt.Println(err)
 		os.Exit(1)
@@ -63,22 +75,36 @@ func main() {
 	queue := liteq.New(liteqDB)
 	// The queue processes as long as this context is not cancelled.
 
-	log.Println("Setting up workers...")
+	log.Println("Setting up worker queues...")
-	go runWorkers(queue)
+	queues := []struct {
+		queueName string
+		worker    queueWorker
+	}{
+		{"fetch", Fetch},
+	}
+	for _, q := range queues {
+		go runQ(queue, q.queueName, q.worker)
+	}
+	return queue
+}
+
+func main() {
+	// Don't let `main()` exit
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+
+	queue := setupLiteQ()
 
-	log.Println("Building network...")
 	// Create the network for the search engine.
 	chUrl := make(chan string)
 
 	go Entre(queue, chUrl)
-	for {
+	for range 5 {
 		chUrl <- "https://jadud.com/"
-		time.Sleep(2 * time.Second)
 		chUrl <- "https://berea.us/"
-		time.Sleep(2 * time.Second)
 	}
 
 	// Don't exit.
-	log.Println("Waiting...")
+	log.Println("Waiting for godot...")
 	wg.Wait()
 }

go.mod

@@ -1,4 +1,4 @@
-module git.jadud.com/grosbeak
+module git.jadud.com/jadudm/grosbeak
 
 go 1.24.0
 

internal/domain64/database.go

@@ -8,7 +8,7 @@ import (
 	"slices"
 	"sync"
 
-	sqlc "git.jadud.com/grosbeak/internal/domain64/sqlc"
+	sqlc "git.jadud.com/jadudm/grosbeak/internal/domain64/sqlc"
 	"github.com/jpillora/go-tld"
 )
 

internal/liteq/bench/main.go

@@ -7,7 +7,7 @@ import (
 	"math/rand"
 	"time"
 
-	liteq "git.jadud.com/grosbeak/internal/liteq"
+	liteq "git.jadud.com/jadudm/grosbeak/internal/liteq"
 	_ "modernc.org/sqlite"
 )
 

internal/liteq/db/queries.sql

@@ -20,10 +20,9 @@ VALUES
         unixepoch(),
         ?,
         ?
-    ) ON CONFLICT (deduping_key, job_status)
+    ) ON CONFLICT (deduping_key)
 WHERE
-    deduping_key != ''
+    deduping_key != '' DO NOTHING;
-    AND (job_status = 'queued' OR job_status = 'fetched' OR job_status = 'completed') DO NOTHING;
 
 -- name: doQueueJobReplaceDupe :exec
 INSERT INTO

internal/liteq/db/schema.sql

@@ -21,10 +21,9 @@ WHERE
     job_status = 'queued'
     OR job_status = 'fetched';
 
-CREATE UNIQUE INDEX IF NOT EXISTS dedupe_ignore ON jobs (deduping_key, job_status)
+CREATE UNIQUE INDEX IF NOT EXISTS dedupe_ignore ON jobs (deduping_key)
 WHERE
-    deduping_key != ''
+    deduping_key != '';
-    AND (job_status = 'queued' OR job_status = 'fetched' OR job_status = 'completed');
 
 CREATE UNIQUE INDEX IF NOT EXISTS dedupe_replace ON jobs (deduping_key, job_status)
 WHERE

internal/liteq/internal/jobs_test.go

@@ -10,7 +10,7 @@ import (
 	"testing"
 	"time"
 
-	"git.jadud.com/grosbeak/internal/liteq/internal"
+	"git.jadud.com/jadudm/grosbeak/internal/liteq/internal"
 	"github.com/matryer/is"
 	_ "modernc.org/sqlite"
 )

internal/liteq/internal/methods.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"log"
 	"time"
 
 	"database/sql/driver"
@@ -110,6 +111,7 @@ func (q *Queries) Consume(ctx context.Context, params ConsumeParams) error {
 	for {
 		// If the context gets canceled for example, stop consuming
 		if ctx.Err() != nil {
+			log.Println("context cancelled in Consume()")
 			return nil
 		}
 
@@ -120,6 +122,7 @@ func (q *Queries) Consume(ctx context.Context, params ConsumeParams) error {
 			})
 
 			if err != nil {
+				log.Println("error resetting jobs", err.Error())
 				return fmt.Errorf("error resetting jobs: %w", err)
 			}
 		}
@@ -143,6 +146,7 @@ func (q *Queries) Consume(ctx context.Context, params ConsumeParams) error {
 			workers.Submit(func() {
 				err := params.Worker(ctx, job)
 				if err != nil {
+					log.Println("worker failled", err.Error())
 					q.FailJob(ctx, FailJobParams{
 						ID:     job.ID,
 						Errors: ErrorList(append(job.Errors, err.Error())),

internal/liteq/internal/queries.sql.go

@@ -205,10 +205,9 @@ VALUES
         unixepoch(),
         ?,
         ?
-    ) ON CONFLICT (deduping_key, job_status)
+    ) ON CONFLICT (deduping_key)
 WHERE
-    deduping_key != ''
+    deduping_key != '' DO NOTHING
-    AND (job_status = 'queued' OR job_status = 'fetched' OR job_status = 'completed') DO NOTHING
 `
 
 type doQueueJobIgnoreDupeParams struct {

internal/liteq/internal/schema.go

@@ -25,10 +25,9 @@ WHERE
     job_status = 'queued'
     OR job_status = 'fetched';
 
-CREATE UNIQUE INDEX IF NOT EXISTS dedupe_ignore ON jobs (deduping_key, job_status)
+CREATE UNIQUE INDEX IF NOT EXISTS dedupe_ignore ON jobs (deduping_key)
 WHERE
-    deduping_key != ''
+    deduping_key != '';
-    AND (job_status = 'queued' OR job_status = 'fetched' OR job_status = 'completed');
 
 CREATE UNIQUE INDEX IF NOT EXISTS dedupe_replace ON jobs (deduping_key, job_status)
 WHERE

internal/liteq/liteq.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"database/sql"
 
-	internal "git.jadud.com/grosbeak/internal/liteq/internal"
+	internal "git.jadud.com/jadudm/grosbeak/internal/liteq/internal"
 )
 
 // Creates the db file with the tables and indexes