Skip to content

feat: temporal tasks scheduler #37

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat: temporal tasks scheduler #37

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions .claude/settings.local.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,11 @@
"Bash(rm:*)",
"Bash(make:*)",
"Bash(timeout 5s make run)",
"Bash(deadcode:*)"
"Bash(deadcode:*)",
"Bash(sqlc generate)",
"Bash(pg_isready -h localhost -p 5432)",
"Bash(psql -h localhost -p 5432 -U postgres -c \"SELECT 1\")"
],
"deny": []
}
}
}
26 changes: 26 additions & 0 deletions cmd/server/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,9 @@ import (
"github.com/eternisai/enchanted-proxy/internal/request_tracking"
"github.com/eternisai/enchanted-proxy/internal/search"
"github.com/eternisai/enchanted-proxy/internal/storage/pg"
"github.com/eternisai/enchanted-proxy/internal/tasks"
"github.com/eternisai/enchanted-proxy/internal/telegram"
"github.com/eternisai/enchanted-proxy/internal/temporal"
"github.com/gin-gonic/gin"
"github.com/go-chi/chi/v5"
"github.com/gorilla/websocket"
Expand Down Expand Up @@ -87,6 +89,19 @@ func main() {
}
log.Info("database connection established")

// Initialize Temporal client
log.Info("initializing temporal client",
slog.String("address", config.AppConfig.TemporalAddress),
slog.String("namespace", config.AppConfig.TemporalNamespace),
slog.String("task_queue", config.AppConfig.TemporalTaskQueue),
)
temporalClient, err := temporal.NewTemporalClientFromConfig(config.AppConfig)
if err != nil {
log.Error("failed to initialize temporal client", slog.String("error", err.Error()))
os.Exit(1)
}
log.Info("temporal client initialized")

Comment on lines +92 to +104
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion | 🟠 Major

Close the Temporal client on shutdown to avoid leaking the gRPC connection.

Add a deferred close right after successful init.

Apply this diff:

  temporalClient, err := temporal.NewTemporalClientFromConfig(config.AppConfig)
  if err != nil {
    log.Error("failed to initialize temporal client", slog.String("error", err.Error()))
    os.Exit(1)
  }
  log.Info("temporal client initialized")
+ // Ensure Temporal client is closed on shutdown.
+ defer func() {
+   if err := temporalClient.Close(); err != nil {
+     log.Warn("temporal client close error", slog.String("error", err.Error()))
+   }
+ }()
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
// Initialize Temporal client
log.Info("initializing temporal client",
slog.String("address", config.AppConfig.TemporalAddress),
slog.String("namespace", config.AppConfig.TemporalNamespace),
slog.String("task_queue", config.AppConfig.TemporalTaskQueue),
)
temporalClient, err := temporal.NewTemporalClientFromConfig(config.AppConfig)
if err != nil {
log.Error("failed to initialize temporal client", slog.String("error", err.Error()))
os.Exit(1)
}
log.Info("temporal client initialized")
// Initialize Temporal client
log.Info("initializing temporal client",
slog.String("address", config.AppConfig.TemporalAddress),
slog.String("namespace", config.AppConfig.TemporalNamespace),
slog.String("task_queue", config.AppConfig.TemporalTaskQueue),
)
temporalClient, err := temporal.NewTemporalClientFromConfig(config.AppConfig)
if err != nil {
log.Error("failed to initialize temporal client", slog.String("error", err.Error()))
os.Exit(1)
}
log.Info("temporal client initialized")
// Ensure Temporal client is closed on shutdown.
defer func() {
if err := temporalClient.Close(); err != nil {
log.Warn("temporal client close error", slog.String("error", err.Error()))
}
}()
🤖 Prompt for AI Agents 
In cmd/server/main.go around lines 92 to 104, the Temporal client is created but
not closed on shutdown, leaking the gRPC connection; after temporalClient is
successfully initialized, add a deferred call to close it (e.g., defer func() {
if err := temporalClient.Close(); err != nil { log.Error("error closing temporal
client", slog.String("error", err.Error())) } }()) so the client is closed when
main exits and any close error is logged.

tokenValidator, err := NewTokenValidator(config.AppConfig, logger)
if err != nil {
log.Error("failed to initialize token validator", slog.String("error", err.Error()))
Expand Down Expand Up @@ -139,6 +154,7 @@ func main() {
iapHandler := iap.NewHandler(iapService, logger.WithComponent("iap"))
mcpHandler := mcp.NewHandler(mcpService)
searchHandler := search.NewHandler(searchService, logger.WithComponent("search"))
tasksHandler := tasks.NewHandler(db.TasksQueries, temporalClient)

// Initialize NATS for Telegram
var natsClient *nats.Conn
Expand Down Expand Up @@ -193,6 +209,7 @@ func main() {
iapHandler: iapHandler,
mcpHandler: mcpHandler,
searchHandler: searchHandler,
tasksHandler: tasksHandler,
deeprStorage: deeprStorage,
})

Expand Down Expand Up @@ -294,6 +311,7 @@ type restServerInput struct {
iapHandler *iap.Handler
mcpHandler *mcp.Handler
searchHandler *search.Handler
tasksHandler *tasks.Handler
deeprStorage deepr.MessageStorage
}

Expand Down Expand Up @@ -368,6 +386,14 @@ func setupRESTServer(input restServerInput) *gin.Engine {
api.POST("/search", input.searchHandler.PostSearchHandler) // POST /api/v1/search (SerpAPI)
api.POST("/exa/search", input.searchHandler.PostExaSearchHandler) // POST /api/v1/exa/search (Exa AI)

// Tasks API routes (protected)
tasksGroup := api.Group("/tasks")
{
tasksGroup.POST("", input.tasksHandler.CreateTask) // POST /api/v1/tasks
tasksGroup.GET("", input.tasksHandler.GetTasks) // GET /api/v1/tasks
tasksGroup.DELETE("/:id", input.tasksHandler.DeleteTask) // DELETE /api/v1/tasks/:id
}

// Deep Research WebSocket endpoint (protected)
api.GET("/deepresearch/ws", deepr.DeepResearchHandler(input.logger, input.requestTrackingService, input.firebaseClient, input.deeprStorage)) // WebSocket proxy for deep research
}
Expand Down
17 changes: 15 additions & 2 deletions go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ module github.com/eternisai/enchanted-proxy
go 1.24.2

require (
cloud.google.com/go/firestore v1.18.0
firebase.google.com/go/v4 v4.16.1
github.com/99designs/gqlgen v0.17.76
github.com/gin-gonic/gin v1.9.1
Expand All @@ -22,7 +23,9 @@ require (
github.com/rs/cors v1.11.1
github.com/sirupsen/logrus v1.9.3
github.com/vektah/gqlparser/v2 v2.5.30
go.temporal.io/sdk v1.37.0
google.golang.org/api v0.231.0
google.golang.org/grpc v1.72.0
)

require (
Expand All @@ -31,7 +34,6 @@ require (
cloud.google.com/go/auth v0.16.1 // indirect
cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect
cloud.google.com/go/compute/metadata v0.6.0 // indirect
cloud.google.com/go/firestore v1.18.0 // indirect
cloud.google.com/go/iam v1.5.2 // indirect
cloud.google.com/go/longrunning v0.6.7 // indirect
cloud.google.com/go/monitoring v1.24.2 // indirect
Expand All @@ -48,9 +50,11 @@ require (
github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311 // indirect
github.com/cncf/xds/go v0.0.0-20250501225837-2ac532fd4443 // indirect
github.com/cpuguy83/go-md2man/v2 v2.0.7 // indirect
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 // indirect
github.com/envoyproxy/go-control-plane/envoy v1.32.4 // indirect
github.com/envoyproxy/protoc-gen-validate v1.2.1 // indirect
github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a // indirect
github.com/felixge/httpsnoop v1.0.4 // indirect
github.com/gabriel-vasile/mimetype v1.4.2 // indirect
github.com/gin-contrib/sse v0.1.0 // indirect
Expand All @@ -62,11 +66,15 @@ require (
github.com/go-playground/validator/v10 v10.14.0 // indirect
github.com/go-viper/mapstructure/v2 v2.3.0 // indirect
github.com/goccy/go-json v0.10.3 // indirect
github.com/gogo/protobuf v1.3.2 // indirect
github.com/golang-jwt/jwt/v5 v5.2.2 // indirect
github.com/golang/mock v1.6.0 // indirect
github.com/golang/protobuf v1.5.4 // indirect
github.com/google/s2a-go v0.1.9 // indirect
github.com/googleapis/enterprise-certificate-proxy v0.3.6 // indirect
github.com/googleapis/gax-go/v2 v2.14.1 // indirect
github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.2 // indirect
github.com/grpc-ecosystem/grpc-gateway/v2 v2.22.0 // indirect
github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
github.com/json-iterator/go v1.1.12 // indirect
github.com/klauspost/compress v1.18.0 // indirect
Expand All @@ -84,14 +92,19 @@ require (
github.com/modern-go/reflect2 v1.0.2 // indirect
github.com/nats-io/nkeys v0.4.11 // indirect
github.com/nats-io/nuid v1.0.1 // indirect
github.com/nexus-rpc/sdk-go v0.3.0 // indirect
github.com/pelletier/go-toml/v2 v2.0.8 // indirect
github.com/pkg/errors v0.9.1 // indirect
github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect
github.com/robfig/cron v1.2.0 // indirect
github.com/russross/blackfriday/v2 v2.1.0 // indirect
github.com/sethvargo/go-retry v0.3.0 // indirect
github.com/sosodev/duration v1.3.1 // indirect
github.com/spf13/cast v1.7.1 // indirect
github.com/spiffe/go-spiffe/v2 v2.5.0 // indirect
github.com/stretchr/objx v0.5.2 // indirect
github.com/stretchr/testify v1.10.0 // indirect
github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
github.com/ugorji/go/codec v1.2.11 // indirect
github.com/urfave/cli/v2 v2.27.7 // indirect
Expand All @@ -108,6 +121,7 @@ require (
go.opentelemetry.io/otel/sdk v1.35.0 // indirect
go.opentelemetry.io/otel/sdk/metric v1.35.0 // indirect
go.opentelemetry.io/otel/trace v1.35.0 // indirect
go.temporal.io/api v1.53.0 // indirect
go.uber.org/multierr v1.11.0 // indirect
golang.org/x/arch v0.3.0 // indirect
golang.org/x/crypto v0.39.0 // indirect
Expand All @@ -123,7 +137,6 @@ require (
google.golang.org/genproto v0.0.0-20250505200425-f936aa4a68b2 // indirect
google.golang.org/genproto/googleapis/api v0.0.0-20250505200425-f936aa4a68b2 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20250505200425-f936aa4a68b2 // indirect
google.golang.org/grpc v1.72.0 // indirect
google.golang.org/protobuf v1.36.6 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
)
Loading