Introduces Background Worker for Async Processing

Procfile

@@ -0,0 +1,2 @@
+web: ./bin/web
+worker: ./bin/worker

README.md

@@ -1,20 +1,54 @@
 # Codex Backend
 
-Backend for Codex - novel reader app.
+Backend for Codex - novel reading platform.
 
-# Details
+## Details
 
-Codex-Backend is built in `GoLang`, using `Gin` for server and `AWS-dynamoDB` for database.
+Codex-Backend is built in `GoLang`, using `Gin` for server and ~AWS-dynamoDB~ firestore (moving to Heroku Postgres) for database.
 
 It is deployed on `Heroku` (thats why the code is in api directory).
 
-### Endpoints
+air config is outdated and not recommended. use [Run](Run guide instead)
 
-This will be updated later (the version before was already outdated)
+## Run
+run server:
 
-### Run
+```bash
+go run api/cmd/web/main.go
+```
 
-- run server:
-    ```bash
-    go run api/cmd/codex/main.go
-    ```
+```bash
+go run api/cmd/worker/main.go
+```
+
+Both are needed
+
+## Endpoints
+
+3 Groups of endpoints: Client, Manage and User.
+
+- Client is responsible for basic GET requests.
+- Manage is responsible for Upload/Modification operations.
+- User is responsible for user authentication, authorization and Registration (Delete is not yet implemented).
+
+### Client: base path followed by request path
+- `/all` - Get all novels
+- `/:novel` - Get a novel by id
+- `/:novel/:chapter` - Get chapter from novel using both ids
+- `/:novel/all` - Get all chapters from novel using id
+- `/:novel/chapter` - Get cursor paginated chapters from novel using id
+
+    Options: limit (max 100), cursor (chapter index (integer)) and sort ("asc" || "desc").
+
+    Defaults: limit=100, cursor=0, sort="desc"
+
+### Manage: `/manage` followed by request path
+- `/upload` - Upload novel
+- `/:novel` - Update novel
+- `/:novel/:chapter` - Update chapter
+
+### User: `/user` followed by request path
+- `/validate` - Validate user token
+- `/login` - Login user
+- `/register` - Register user
+- `/logout` - Logout user

api/cmd/web/main.go

@@ -0,0 +1,19 @@
+package main
+
+import (
+	cmn "Codex-Backend/api/internal/common"
+	firestore_server "Codex-Backend/api/internal/interfaces/rest"
+	"os"
+
+	_ "github.com/heroku/x/hmetrics/onload"
+)
+
+func init() {
+	if mode := os.Getenv("GIN_MODE"); mode == "debug" {
+		cmn.LoadEnvVariables()
+	}
+}
+
+func main() {
+	firestore_server.Server()
+}

api/cmd/worker/main.go

@@ -0,0 +1,47 @@
+package main
+
+import (
+	cmn "Codex-Backend/api/internal/common"
+	queue "Codex-Backend/api/internal/common/river"
+	"context"
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+)
+
+func init() {
+	if mode := os.Getenv("GIN_MODE"); mode == "debug" {
+		cmn.LoadEnvVariables()
+	}
+}
+
+func main() {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	riverClient := queue.GetRiverClient(ctx)
+	if err := riverClient.Start(ctx); err != nil {
+		log.Fatal("Failed to start River client:", err)
+	}
+
+	log.Println("River worker started successfully, waiting for jobs...")
+
+	sigChan := make(chan os.Signal, 1)
+	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
+
+	<-sigChan
+	log.Println("Shutdown signal received, starting graceful shutdown...")
+
+	cancel()
+
+	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer shutdownCancel()
+
+	if err := riverClient.Stop(shutdownCtx); err != nil {
+		log.Printf("Error during River client shutdown: %v", err)
+	} else {
+		log.Println("River worker stopped gracefully")
+	}
+}

api/internal/common/env.go

@@ -2,21 +2,25 @@ package common
 
 import (
 	"errors"
+	"log"
 	"net/http"
 	"os"
 
 	"github.com/joho/godotenv"
 )
 
-func LoadEnvVariables() error {
-	return godotenv.Load(".env")
+func LoadEnvVariables() {
+	err := godotenv.Load(".env")
+	if err != nil {
+		log.Fatal(&Error{Err: errors.New("Failed to load environment variables"), Status: http.StatusInternalServerError})
+	}
 }
 
-func GetEnvVariable(v string) (string, error) {
+func GetEnvVariable(v string) string {
 	env_variable := os.Getenv(v)
 	if env_variable == "" {
-		return "", &Error{Err: errors.New("Environmental Variable " + v + " Not Found"), Status: http.StatusNotFound}
+		log.Fatal(&Error{Err: errors.New("Environmental Variable " + v + " Not Found"), Status: http.StatusNotFound})
 	}
 
-	return env_variable, nil
+	return env_variable
 }

api/internal/common/river/client.go

@@ -0,0 +1,60 @@
+package queue
+
+import (
+	"Codex-Backend/api/internal/usecases/worker"
+	"context"
+	"log"
+	"log/slog"
+	"os"
+	"sync"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/riverqueue/river"
+	"github.com/riverqueue/river/riverdriver/riverpgxv5"
+)
+
+var (
+	riverClient *river.Client[pgx.Tx]
+	riverOnce   sync.Once
+)
+
+func InitializeRiverClient(ctx context.Context, workers *river.Workers) *river.Client[pgx.Tx] {
+	dbPool, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
+	if err != nil {
+		panic(err)
+	}
+
+	logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
+		Level: slog.LevelInfo,
+	}))
+
+	riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
+		Logger: logger,
+		Queues: map[string]river.QueueConfig{
+			river.QueueDefault: {MaxWorkers: 10},
+		},
+		MaxAttempts: 3,
+		Workers:     workers,
+	})
+	if err != nil {
+		panic(err)
+	}
+
+	return riverClient
+}
+
+func GetRiverClient(ctx context.Context) *river.Client[pgx.Tx] {
+	riverOnce.Do(func() {
+		log.Println("Initializing River client...")
+
+		workers := river.NewWorkers()
+		river.AddWorker(workers, &worker.EPUBWorker{})
+
+		riverClient = InitializeRiverClient(ctx, workers)
+
+		log.Println("River client initialized successfully")
+	})
+
+	return riverClient
+}

api/internal/common/token.go

@@ -10,10 +10,7 @@ import (
 
 func GenerateToken(email string) (string, error) {
 
-	key, err := GetEnvVariable("JWT_SIGN_KEY")
-	if err != nil {
-		return "", err
-	}
+	key := GetEnvVariable("JWT_SIGN_KEY")
 	signKey := []byte(key)
 
 	expirationTime := time.Now().Add(time.Hour * 24)

api/internal/infrastructure/client/client.go

@@ -18,10 +18,7 @@ type Client struct {
 func FirestoreClient() (*firestore.Client, error) {
 	ctx := context.Background()
 
-	credentials_json, err := cmn.GetEnvVariable("GOOGLE_CREDENTIALS")
-	if err != nil {
-		return nil, &cmn.Error{Err: errors.New("Firestore Client Error - GOOGLE_CREDENTIALS: " + err.Error()), Status: http.StatusInternalServerError}
-	}
+	credentials_json := cmn.GetEnvVariable("GOOGLE_CREDENTIALS")
 
 	sa := option.WithCredentialsJSON([]byte(credentials_json))
 

api/internal/infrastructure/collections/chapters.go

@@ -21,30 +21,16 @@ func (c *Client) CursorPagination(options domain.CursorOptions, ctx context.Cont
 	limit := min(max(options.Limit, 1), 100)
 
 	snapshots := []*firestore.DocumentSnapshot{}
+	var err error
 
 	if options.Cursor == 0 {
-		snaps, err := query.Limit(1).Documents(ctx).GetAll()
-		if err != nil {
-			return nil, err
-		}
-
-		if len(snaps) == 0 {
-			return nil, &cmn.Error{
-				Err:    fmt.Errorf("Firestore Client Error - Get Paginated Chapters - No Chapters Found for Novel: %s", options.NovelID),
-				Status: http.StatusNotFound,
-			}
-		}
-
-		snapshots, err = query.StartAt(snaps[0]).Limit(limit + 1).Documents(ctx).GetAll()
-		if err != nil {
-			return nil, err
-		}
+		snapshots, err = query.Limit(limit + 1).Documents(ctx).GetAll()
 	} else {
-		var err error
 		snapshots, err = query.StartAt(options.Cursor).Limit(limit + 1).Documents(ctx).GetAll()
-		if err != nil {
-			return nil, err
-		}
+	}
+
+	if err != nil {
+		return nil, err
 	}
 
 	if len(snapshots) == 0 {
@@ -54,22 +40,10 @@ func (c *Client) CursorPagination(options domain.CursorOptions, ctx context.Cont
 		}
 	}
 
-	nextCursor := 0
-	if len(snapshots) > limit {
-		var chapter domain.Chapter
-		if err := snapshots[len(snapshots)-1].DataTo(&chapter); err != nil {
-			return nil, err
-		}
-		nextCursor = chapter.Index
-	}
-
-	snapLen := len(snapshots) - 1
-	if snapLen <= 0 {
-		snapLen++
-	}
+	actualLimit := min(len(snapshots), limit)
+	chapters := make([]domain.FrontendChapter, 0, actualLimit)
 
-	chapters := []domain.FrontendChapter{}
-	for _, snapshot := range snapshots[:snapLen] {
+	for _, snapshot := range snapshots[:actualLimit] {
 		var chapter domain.Chapter
 		if err := snapshot.DataTo(&chapter); err != nil {
 			return nil, err
@@ -82,6 +56,15 @@ func (c *Client) CursorPagination(options domain.CursorOptions, ctx context.Cont
 		})
 	}
 
+	nextCursor := 0
+	if len(snapshots) > limit {
+		var lastChapter domain.Chapter
+		if err := snapshots[limit].DataTo(&lastChapter); err != nil {
+			return nil, err
+		}
+		nextCursor = lastChapter.Index
+	}
+
 	return &domain.CursorResponse{
 		Chapters:   chapters,
 		NextCursor: nextCursor,
@@ -95,7 +78,7 @@ func (c *Client) BatchUploadChapters(novelId string, chapters []domain.Chapter,
 	for i := 0; i < len(chapters); i += chunkSize {
 		subset := chapters[i:min(i+chunkSize, len(chapters))]
 
-		batchCtx, cancel := context.WithTimeout(ctx, 300*time.Second)
+		batchCtx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
 		defer cancel()
 
 		bw := c.Client.BulkWriter(batchCtx)
@@ -104,6 +87,7 @@ func (c *Client) BatchUploadChapters(novelId string, chapters []domain.Chapter,
 		for _, chap := range subset {
 			job, err := bw.Set(coll.Doc(chap.ID), chap)
 			if err != nil {
+				cancel()
 				return &cmn.Error{
 					Err:    fmt.Errorf("Firestore Client Error - Batch Upload Chapters - Enqueue failed for chapter %s: %w", chap.ID, err),
 					Status: http.StatusInternalServerError,
@@ -119,14 +103,18 @@ func (c *Client) BatchUploadChapters(novelId string, chapters []domain.Chapter,
 		for j, job := range jobs {
 			if _, err := job.Results(); err != nil {
 				chap := subset[j]
+				cancel()
 				return &cmn.Error{
 					Err:    fmt.Errorf("Firestore Client Error - Batch Upload Chapters - Write failed for chapter %s: %w", chap.ID, err),
 					Status: http.StatusInternalServerError,
 				}
 			}
 		}
 
-		time.Sleep(100 * time.Millisecond)
+		cancel()
+		if i+chunkSize < len(chapters) {
+			time.Sleep(200 * time.Millisecond)
+		}
 	}
 
 	return nil