added admin app: fixes

Author: absorbb

admin/failover_reprocessor.go

@@ -5,9 +5,6 @@ import (
 	"compress/gzip"
 	"context"
 	"fmt"
-	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
-	"github.com/jitsucom/bulker/jitsubase/jsonorder"
-	"github.com/jitsucom/bulker/jitsubase/utils"
 	"io"
 	"os"
 	"path/filepath"
@@ -18,6 +15,10 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
+	"github.com/jitsucom/bulker/jitsubase/jsonorder"
+	"github.com/jitsucom/bulker/jitsubase/utils"
+
 	"github.com/aws/aws-sdk-go-v2/aws"
 	awsconfig "github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
@@ -323,7 +324,7 @@ func (m *ReprocessingJobManager) processJob(job *ReprocessingJob) {
 			m.Infof("Reached event limit of %d, stopping processing", job.Config.Limit)
 			break
 		}
-		
+
 		select {
 		case <-job.ctx.Done():
 			return
@@ -493,13 +494,13 @@ func (m *ReprocessingJobManager) processFile(job *ReprocessingJob, filePath stri
 		}
 
 		lineNum++
-		
+
 		// Check if we've reached the limit
 		if job.Config.Limit > 0 && atomic.LoadInt64(&job.SuccessCount) >= job.Config.Limit {
 			m.Infof("Reached event limit of %d, stopping processing", job.Config.Limit)
 			break
 		}
-		
+
 		line := scanner.Bytes()
 		// Parse message as IngestMessage
 		var ingestMsg IngestMessage
@@ -564,7 +565,8 @@ func (m *ReprocessingJobManager) filterMessage(job *ReprocessingJob, ingestMsg *
 
 	// Filter by streamIds if specified
 	if len(job.Config.StreamIds) > 0 {
-		return utils.ArrayContains(job.Config.StreamIds, ingestMsg.Origin.SourceId)
+		return utils.ArrayContains(job.Config.StreamIds, ingestMsg.Origin.SourceId) ||
+			utils.ArrayContains(job.Config.StreamIds, ingestMsg.Origin.Slug)
 	}
 
 	return true
@@ -583,7 +585,7 @@ func (m *ReprocessingJobManager) processBatch(job *ReprocessingJob, batch []*Ing
 			batch = batch[:remaining]
 		}
 	}
-	
+
 	if job.Config.DryRun {
 		// In dry run mode, just count the messages
 		atomic.AddInt64(&job.SuccessCount, int64(len(batch)))

admin/router.go

@@ -3,6 +3,8 @@ package main
 import (
 	"github.com/gin-gonic/gin"
 	"github.com/jitsucom/bulker/jitsubase/appbase"
+	"github.com/jitsucom/bulker/jitsubase/utils"
+
 	"net/http"
 )
 
@@ -15,7 +17,7 @@ type Router struct {
 func NewRouter(context *Context) *Router {
 	base := appbase.NewRouterBase(context.config.Config, []string{
 		"/health",
-		"/admin",
+		"/",
 		"/static",
 	})
 	router := &Router{
@@ -40,9 +42,30 @@ func NewRouter(context *Context) *Router {
 	reprocessingAPI.POST("/jobs/:id/cancel", router.cancelReprocessingJob)
 
 	// Serve HTML interface for admin (no auth required for UI, auth handled via form)
-	engine.GET("/admin", router.serveAdminHTML)
+	engine.GET("/", router.serveAdminHTML)
 	engine.Static("/static", "./static")
 
 	return router
 
 }
+func (r *Router) CorsMiddleware(c *gin.Context) {
+	origin := c.GetHeader("Origin")
+	if c.Request.Method == "OPTIONS" {
+		c.Header("Access-Control-Allow-Origin", utils.NvlString(origin, "*"))
+		c.Header("Access-Control-Allow-Methods", "GET,POST,HEAD,OPTIONS")
+		// x-jitsu-custom - in case client want to add some custom payload via header
+		c.Header("Access-Control-Allow-Headers", "x-enable-debug, x-write-key, authorization, content-type, x-ip-policy, cache-control, x-jitsu-custom")
+		c.Header("Access-Control-Allow-Credentials", "true")
+		c.Header("Access-Control-Max-Age", "86400")
+		c.AbortWithStatus(http.StatusOK)
+		return
+	} else if origin != "" {
+		c.Header("Access-Control-Allow-Origin", origin)
+		c.Header("Access-Control-Allow-Methods", "GET,POST,HEAD,OPTIONS")
+		// x-jitsu-custom - in case client want to add some custom payload via header
+		c.Header("Access-Control-Allow-Headers", "x-enable-debug, x-write-key, authorization, content-type, x-ip-policy, cache-control, x-jitsu-custom")
+		c.Header("Access-Control-Allow-Credentials", "true")
+		c.Header("Access-Control-Max-Age", "86400")
+	}
+	c.Next()
+}