Disable logging and switch event callback writing buffer check

inngest · Oct 8, 2024 · fbe5747 · fbe5747
1 parent f84d07e
commit fbe5747
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 10 deletions.
diff --git a/internal/test/pg_data.go b/internal/test/pg_data.go
@@ -17,7 +17,7 @@ import (
 
 const (
 	DefaultSeed        = 123
-	DefaultAccountUUID = "6db2bd8a-2a2f-52d3-aa79-abb4015d6dbd"
+	DefaultAccountUUID = "9b332174-2fc5-5781-8aba-b2500384cc1c"
 	DefaultEmail       = "lriai1h2oy1d@example.com"
 )
 
@@ -89,10 +89,6 @@ func InsertAccounts(t *testing.T, ctx context.Context, cfg pgx.ConnConfig, opts
 	}
 }
 
-func InsertAccountsAndUsers(t *testing.T, ctx context.Context, opts InsertOpts) {
-	panic("nah")
-}
-
 func hash(in any) string {
 	switch v := in.(type) {
 	case string:

diff --git a/internal/test/pg_init.go b/internal/test/pg_init.go
@@ -3,9 +3,11 @@ package test
 import (
 	"context"
 	"fmt"
+	"log"
 	"strings"
 	"testing"
 
+	"github.com/docker/docker/pkg/ioutils"
 	"github.com/inngest/dbcap/pkg/replicator/pgreplicator/pgsetup"
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgconn"
@@ -24,6 +26,10 @@ type StartPGOpts struct {
 	DisableReplicaIdentityFull bool
 }
 
+func init() {
+	tc.Logger = log.New(&ioutils.NopWriter{}, "", 0)
+}
+
 func StartPG(t *testing.T, ctx context.Context, opts StartPGOpts) (tc.Container, pgx.ConnConfig) {
 	t.Helper()
 	args := []tc.ContainerCustomizer{

diff --git a/pkg/eventwriter/callback_writer.go b/pkg/eventwriter/callback_writer.go
@@ -82,6 +82,10 @@ func (a *cbWriter) Listen(ctx context.Context, committer changeset.WatermarkComm
 				buf = make([]*changeset.Changeset, a.batchSize)
 				i = 0
 			case msg := <-a.cs:
+				buf[i] = msg
+				i++
+				// Send this batch after at least 5 seconds
+				timer.Reset(a.batchTimeout)
 				if i == a.batchSize {
 					// send this batch, as we're full.
 					if err := a.onChangeset(buf); err == nil {
@@ -92,11 +96,6 @@ func (a *cbWriter) Listen(ctx context.Context, committer changeset.WatermarkComm
 					i = 0
 					continue
 				}
-				// Appoend the
-				buf[i] = msg
-				i++
-				// Send this batch after at least 5 seconds
-				timer.Reset(a.batchTimeout)
 			}
 		}
 	}()