feat: Start writing the Routing API

.env

@@ -2,3 +2,4 @@ DATABASE_URL="user=dionysia password=dionysia dbname=dionysia host=db port=5432"
 REDIS_ADDR=redis:6379
 API_PORT=8080
 READ_HEADER_TIMEOUT=2s
+ORIGIN_TTL=10

cmd/api.go

@@ -6,6 +6,7 @@ import (
 	"github.com/dionysia-dev/dionysia/internal/api"
 	"github.com/dionysia-dev/dionysia/internal/config"
 	"github.com/dionysia-dev/dionysia/internal/db"
+	"github.com/dionysia-dev/dionysia/internal/db/redistore"
 	"github.com/dionysia-dev/dionysia/internal/logging"
 	"github.com/dionysia-dev/dionysia/internal/queue"
 	"github.com/dionysia-dev/dionysia/internal/service"
@@ -30,6 +31,12 @@ func NewAPICmd() *cobra.Command {
 					db.New,
 					db.NewDBInputStore,
 					queue.NewClient,
+					redistore.NewClient,
+					fx.Annotate(
+						redistore.NewOriginStore,
+						fx.As(new(service.OriginStore)),
+					),
+					service.NewOriginHandler,
 					service.NewNotificationHandler,
 				),
 				api.Module,

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/hibiken/asynq v0.24.1
 	github.com/joho/godotenv v1.5.1
+	github.com/redis/go-redis/v9 v9.0.3
 	github.com/spf13/cobra v1.8.0
 	github.com/stretchr/testify v1.9.0
 	github.com/swaggo/files v1.0.1
@@ -56,7 +57,6 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/redis/go-redis/v9 v9.0.3 // indirect
 	github.com/robfig/cron/v3 v3.0.1 // indirect
 	github.com/rogpeppe/go-internal v1.12.0 // indirect
 	github.com/spf13/cast v1.3.1 // indirect

internal/api/api.go

@@ -15,17 +15,17 @@ import (
 	swaggerFiles "github.com/swaggo/files"
 	ginSwagger "github.com/swaggo/gin-swagger"
 	"go.uber.org/fx"
-	"go.uber.org/zap"
 )
 
 // @title           Streaming Platform API
 // @version         1.0
 // @description     Manage your streaming workflow
 // @BasePath  /api/v1
 
-func New(inputStore db.InputStore, _ *zap.SugaredLogger, nh service.NotificationHandler) *gin.Engine {
+func New(inputStore db.InputStore, nh service.NotificationHandler, oh service.OriginHandler) *gin.Engine {
 	inputController := NewInputController(service.NewInputHandler(inputStore))
 	notificationController := NewNotificationController(nh)
+	originController := NewOriginController(oh)
 
 	e := gin.Default()
 
@@ -37,6 +37,7 @@ func New(inputStore db.InputStore, _ *zap.SugaredLogger, nh service.Notification
 	v1.DELETE("/inputs/:id", inputController.DeleteInput)
 	v1.POST("/inputs/auth", inputController.Authenticate)
 	v1.POST("/notifications/package", notificationController.EnqueuePackaging)
+	v1.PUT("/origins", originController.UpdateOrigin)
 
 	e.GET("/swagger/*any", ginSwagger.WrapHandler(swaggerFiles.Handler))
 

internal/api/controller.go

@@ -16,6 +16,10 @@ type NotificationController struct {
 	notificationHandler service.NotificationHandler
 }
 
+type OriginController struct {
+	originHandler service.OriginHandler
+}
+
 func NewInputController(inputHandler service.InputHandler) *InputController {
 	return &InputController{inputHandler: inputHandler}
 }
@@ -204,3 +208,38 @@ func (n *NotificationController) EnqueuePackaging(ctx *gin.Context) {
 		Message: "Packaging job enqueued successfully",
 	})
 }
+
+func NewOriginController(oh service.OriginHandler) *OriginController {
+	return &OriginController{originHandler: oh}
+}
+
+func (o *OriginController) UpdateOrigin(ctx *gin.Context) {
+	var originData OriginData
+	if err := ctx.BindJSON(&originData); err != nil {
+		statusCode, response := HandleValidationError(err)
+		if statusCode == 0 && response == nil {
+			ctx.JSON(http.StatusInternalServerError, ErrorResponse{
+				Error: Error{Message: "InternalServerError: failed to update origin"},
+			})
+
+			return
+		}
+
+		ctx.JSON(int(statusCode), response)
+
+		return
+	}
+
+	origin := originData.ToOrigin()
+	if err := o.originHandler.Update(ctx, origin); err != nil {
+		ctx.JSON(http.StatusInternalServerError, ErrorResponse{
+			Error: Error{Message: "InternalServerError: failed to update origin"},
+		})
+
+		return
+	}
+
+	ctx.JSON(http.StatusOK, SuccessResponse{
+		Message: "Origin updated successfully",
+	})
+}

internal/api/request.go

@@ -91,3 +91,15 @@ type IngestAuthData struct {
 	Path   uuid.UUID `json:"path" validate:"required"`
 	Action string    `json:"action" validate:"required"`
 }
+
+type OriginData struct {
+	ID      uuid.UUID `json:"id" validate:"required"`
+	Address string    `json:"address" validate:"required"`
+}
+
+func (o *OriginData) ToOrigin() service.Origin {
+	return service.Origin{
+		ID:      o.ID,
+		Address: o.Address,
+	}
+}

internal/config/config.go

@@ -12,6 +12,7 @@ type Config struct {
 	RedisAddr          string        `env:"REDIS_ADDR" envDefault:"localhost:6379"`
 	MaxConcurrentTasks int           `env:"MAX_CONCURRENT_TASKS" envDefault:"10"`
 	ReadHeaderTimeout  time.Duration `env:"READ_HEADER_TIMEOUT" envDefault:"2"`
+	OriginTTL          int           `env:"ORIGIN_TTL" envDefault:"10"`
 }
 
 func New() (*Config, error) {

internal/db/redistore/client.go

@@ -0,0 +1,12 @@
+package redistore
+
+import (
+	"github.com/dionysia-dev/dionysia/internal/config"
+	"github.com/redis/go-redis/v9"
+)
+
+func NewClient(cfg *config.Config) *redis.Client {
+	return redis.NewClient(&redis.Options{
+		Addr: cfg.RedisAddr,
+	})
+}

internal/db/redistore/store.go

@@ -0,0 +1,56 @@
+package redistore
+
+import (
+	"context"
+
+	"github.com/dionysia-dev/dionysia/internal/config"
+	"github.com/dionysia-dev/dionysia/internal/service"
+	"github.com/google/uuid"
+	"github.com/redis/go-redis/v9"
+)
+
+type OriginStore struct {
+	client *redis.Client
+	ttl    int
+}
+
+func NewOriginStore(client *redis.Client, cfg *config.Config) *OriginStore {
+	return &OriginStore{
+		client: client,
+		ttl:    cfg.OriginTTL,
+	}
+}
+
+func (s *OriginStore) Update(ctx context.Context, origin service.Origin) error {
+	var updateOrigin = redis.NewScript(`
+		local key = KEYS[1]
+		local value = ARGV[1]
+		local ttl = ARGV[2]
+		local curr_value = redis.call("GET", key)
+		if curr_value == value or curr_value == false then
+			redis.call("SETEX", key, ttl, value)
+			return 1
+		end
+		return 0
+	`)
+
+	_, err := updateOrigin.Run(
+		ctx,
+		s.client,
+		[]string{origin.ID.String()}, origin.Address, s.ttl).
+		Result()
+
+	return err
+}
+
+func (s *OriginStore) Get(ctx context.Context, uuid uuid.UUID) (service.Origin, error) {
+	address, err := s.client.Get(ctx, uuid.String()).Result()
+	if err != nil {
+		return service.Origin{}, err
+	}
+
+	return service.Origin{
+		ID:      uuid,
+		Address: address,
+	}, nil
+}

internal/service/routing.go

@@ -0,0 +1,35 @@
+package service
+
+import (
+	"context"
+
+	"github.com/google/uuid"
+)
+
+type OriginStore interface {
+	Update(context.Context, Origin) error
+	Get(context.Context, uuid.UUID) (Origin, error)
+}
+
+type Origin struct {
+	ID      uuid.UUID
+	Address string
+}
+
+type OriginHandler interface {
+	Update(context.Context, Origin) error
+}
+
+type originHandler struct {
+	store OriginStore
+}
+
+func NewOriginHandler(store OriginStore) OriginHandler {
+	return &originHandler{
+		store: store,
+	}
+}
+
+func (h *originHandler) Update(ctx context.Context, origin Origin) error {
+	return h.store.Update(ctx, origin)
+}

internal/service/routing_test.go

@@ -0,0 +1 @@
+package service_test