fix eventhub drop mirror

flow/cmd/handler.go

@@ -8,7 +8,7 @@ import (
 	"github.com/PeerDB-io/peer-flow/generated/protos"
 	"github.com/PeerDB-io/peer-flow/shared"
 	peerflow "github.com/PeerDB-io/peer-flow/workflows"
-	"github.com/cenkalti/backoff"
+	backoff "github.com/cenkalti/backoff/v4"
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5/pgxpool"
 	log "github.com/sirupsen/logrus"
@@ -171,7 +171,7 @@ func (h *FlowRequestHandler) waitForWorkflowClose(ctx context.Context, workflowI
 
 	err := backoff.Retry(operation, expBackoff)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
 	return nil

flow/connectors/eventhub/eventhub.go

@@ -380,12 +380,6 @@ func (c *EventHubConnector) SetupNormalizedTables(
 	}, nil
 }
 
-func (c *EventHubConnector) SyncFlowCleanup(jobName string) error {
-	_, err := c.pgMetadata.pool.Exec(c.ctx, fmt.Sprintf("DROP SCHEMA IF EXISTS %s CASCADE",
-		metadataSchema))
-	return err
-}
-
 func eventDataFromString(s string) *azeventhubs.EventData {
 	return &azeventhubs.EventData{
 		Body: []byte(s),

flow/connectors/eventhub/metadata.go

@@ -226,3 +226,11 @@ func (c *EventHubConnector) incrementSyncBatchID(jobName string) error {
 
 	return nil
 }
+
+func (c *EventHubConnector) SyncFlowCleanup(jobName string) error {
+	_, err := c.pgMetadata.pool.Exec(c.ctx, `
+		DELETE FROM `+metadataSchema+`.`+lastSyncStateTableName+`
+		WHERE job_name = $1
+	`, jobName)
+	return err
+}