diff --git a/pkg/pipe/arbiter.go b/pkg/pipe/arbiter.go
index f28cdd7..3da310f 100644
--- a/pkg/pipe/arbiter.go
+++ b/pkg/pipe/arbiter.go
@@ -4,10 +4,7 @@ import (
 	"context"
 	"fmt"
 	"log"
-	"os"
-	"os/signal"
 	"sync/atomic"
-	"syscall"
 	"time"
 
 	"github.com/vertex-lab/crawler_v2/pkg/graph"
@@ -214,13 +211,3 @@ func Promote(db redb.RedisDB, node graph.ID) error {
 
 	return db.Promote(ctx, node)
 }
-
-// HandleSignals listens for OS signals and triggers context cancellation.
-func HandleSignals(cancel context.CancelFunc) {
-	signals := make(chan os.Signal, 1)
-	signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
-	<-signals
-
-	log.Println("signal received. shutting down...")
-	cancel()
-}
diff --git a/pkg/pipe/intake.go b/pkg/pipe/intake.go
index 555d1d4..9f01bae 100644
--- a/pkg/pipe/intake.go
+++ b/pkg/pipe/intake.go
@@ -5,7 +5,6 @@ import (
 	"errors"
 	"fmt"
 	"log"
-	"slices"
 	"time"
 
 	"github.com/nbd-wtf/go-nostr"
@@ -74,29 +73,6 @@ type PubkeyChecker interface {
 	Exists(ctx context.Context, pubkey string) (bool, error)
 }
 
-// buffer is a minimalistic ring buffer used to keep track of the latest event IDs
-type buffer struct {
-	IDs      []string
-	capacity int
-	write    int
-}
-
-func newBuffer(capacity int) *buffer {
-	return &buffer{
-		IDs:      make([]string, capacity),
-		capacity: capacity,
-	}
-}
-
-func (b *buffer) Add(ID string) {
-	b.IDs[b.write] = ID
-	b.write = (b.write + 1) % b.capacity
-}
-
-func (b *buffer) Contains(ID string) bool {
-	return slices.Contains(b.IDs, ID)
-}
-
 // Firehose connects to a list of relays and pulls config.Kinds events that are newer than [FirehoseConfig.Since].
 // It discards events from unknown pubkeys as an anti-spam mechanism.
 func Firehose(ctx context.Context, config FirehoseConfig, check PubkeyChecker, send func(*nostr.Event) error) {
diff --git a/pkg/pipe/utils.go b/pkg/pipe/utils.go
new file mode 100644
index 0000000..dfaf3f6
--- /dev/null
+++ b/pkg/pipe/utils.go
@@ -0,0 +1,42 @@
+package pipe
+
+import (
+	"context"
+	"log"
+	"os"
+	"os/signal"
+	"slices"
+	"syscall"
+)
+
+// HandleSignals listens for OS signals and triggers context cancellation.
+func HandleSignals(cancel context.CancelFunc) {
+	signals := make(chan os.Signal, 1)
+	signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
+	<-signals
+
+	log.Println("signal received. shutting down...")
+	cancel()
+}
+
+type buffer struct {
+	IDs      []string
+	capacity int
+	write    int
+}
+
+func newBuffer(capacity int) *buffer {
+	return &buffer{
+		IDs:      make([]string, capacity),
+		capacity: capacity,
+	}
+}
+
+func (b *buffer) Add(ID string) {
+	b.IDs[b.write] = ID
+	b.write = (b.write + 1) % b.capacity
+}
+
+func (b *buffer) Contains(ID string) bool {
+	return slices.Contains(b.IDs, ID)
+}