jay/go-heartwood
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix terminology: follower -> traveller

Browse Source
This commit is contained in:
Jay
2026-03-05 08:35:22 -05:00
parent 5f63e4b469
commit ade1a5a10d
2 changed files with 92 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

120
write.go
View File

@@ -16,7 +16,7 @@ type WriteOptions struct {
BoltReadBatchSize int BoltReadBatchSize int
} }
type EventFollower struct { type EventTraveller struct {
ID string ID string
JSON string JSON string
Event roots.Event Event roots.Event
@@ -30,8 +30,8 @@ type WriteResult struct {
} }
type WriteReport struct { type WriteReport struct {
InvalidEvents []EventFollower InvalidEvents []EventTraveller
SkippedEvents []EventFollower SkippedEvents []EventTraveller
CreatedEventCount int CreatedEventCount int
Neo4jResultSummaries []neo4j.ResultSummary Neo4jResultSummaries []neo4j.ResultSummary
Duration time.Duration Duration time.Duration
@@ -58,42 +58,42 @@ func WriteEvents(
var wg sync.WaitGroup var wg sync.WaitGroup
// Create Event Followers // Create Event Travellers
jsonChan := make(chan string) jsonChan := make(chan string)
eventChan := make(chan EventFollower) eventChan := make(chan EventTraveller)
wg.Add(1) wg.Add(1)
go createEventFollowers(&wg, jsonChan, eventChan) go createEventTravellers(&wg, jsonChan, eventChan)
// Parse Event JSON // Parse Event JSON
parsedChan := make(chan EventFollower) parsedChan := make(chan EventTraveller)
invalidChan := make(chan EventFollower) invalidChan := make(chan EventTraveller)
wg.Add(1) wg.Add(1)
go parseEventJSON(&wg, eventChan, parsedChan, invalidChan) go parseEventJSON(&wg, eventChan, parsedChan, invalidChan)
// Collect Invalid Events // Collect Invalid Events
collectedInvalidChan := make(chan []EventFollower) collectedInvalidChan := make(chan []EventTraveller)
wg.Add(1) wg.Add(1)
go collectEvents(&wg, invalidChan, collectedInvalidChan) go collectTravellers(&wg, invalidChan, collectedInvalidChan)
// Enforce Policy Rules // Enforce Policy Rules
queuedChan := make(chan EventFollower) queuedChan := make(chan EventTraveller)
skippedChan := make(chan EventFollower) skippedChan := make(chan EventTraveller)
wg.Add(1) wg.Add(1)
go enforcePolicyRules(&wg, driver, boltdb, opts.BoltReadBatchSize, go enforcePolicyRules(&wg, driver, boltdb, opts.BoltReadBatchSize,
parsedChan, queuedChan, skippedChan) parsedChan, queuedChan, skippedChan)
// Collect Skipped Events // Collect Skipped Events
collectedSkippedChan := make(chan []EventFollower) collectedSkippedChan := make(chan []EventTraveller)
wg.Add(1) wg.Add(1)
go collectEvents(&wg, skippedChan, collectedSkippedChan) go collectTravellers(&wg, skippedChan, collectedSkippedChan)
// Convert Events To Subgraphs // Convert Events To Subgraphs
convertedChan := make(chan EventFollower) convertedChan := make(chan EventTraveller)
wg.Add(1) wg.Add(1)
go convertEventsToSubgraphs(&wg, opts.Expanders, queuedChan, convertedChan) go convertEventsToSubgraphs(&wg, opts.Expanders, queuedChan, convertedChan)
@@ -139,29 +139,29 @@ func setDefaultWriteOptions(opts *WriteOptions) {
} }
} }
func createEventFollowers(wg *sync.WaitGroup, jsonChan chan string, eventChan chan EventFollower) { func createEventTravellers(wg *sync.WaitGroup, jsonChan chan string, eventChan chan EventTraveller) {
defer wg.Done() defer wg.Done()
for json := range jsonChan { for json := range jsonChan {
eventChan <- EventFollower{JSON: json} eventChan <- EventTraveller{JSON: json}
} }
close(eventChan) close(eventChan)
} }
func parseEventJSON(wg *sync.WaitGroup, inChan, parsedChan, invalidChan chan EventFollower) { func parseEventJSON(wg *sync.WaitGroup, inChan, parsedChan, invalidChan chan EventTraveller) {
defer wg.Done() defer wg.Done()
for follower := range inChan { for traveller := range inChan {
var event roots.Event var event roots.Event
jsonBytes := []byte(follower.JSON) jsonBytes := []byte(traveller.JSON)
err := json.Unmarshal(jsonBytes, &event) err := json.Unmarshal(jsonBytes, &event)
if err != nil { if err != nil {
follower.Error = err traveller.Error = err
invalidChan <- follower invalidChan <- traveller
continue continue
} }
follower.ID = event.ID traveller.ID = event.ID
follower.Event = event traveller.Event = event
parsedChan <- follower parsedChan <- traveller
} }
close(parsedChan) close(parsedChan)
@@ -172,17 +172,17 @@ func enforcePolicyRules(
wg *sync.WaitGroup, wg *sync.WaitGroup,
driver neo4j.Driver, boltdb *bolt.DB, driver neo4j.Driver, boltdb *bolt.DB,
batchSize int, batchSize int,
inChan, queuedChan, skippedChan chan EventFollower, inChan, queuedChan, skippedChan chan EventTraveller,
) { ) {
defer wg.Done() defer wg.Done()
var batch []EventFollower var batch []EventTraveller
for follower := range inChan { for traveller := range inChan {
batch = append(batch, follower) batch = append(batch, traveller)
if len(batch) >= batchSize { if len(batch) >= batchSize {
processPolicyRulesBatch(boltdb, batch, queuedChan, skippedChan) processPolicyRulesBatch(boltdb, batch, queuedChan, skippedChan)
batch = []EventFollower{} batch = []EventTraveller{}
} }
} }
@@ -196,35 +196,35 @@ func enforcePolicyRules(
func processPolicyRulesBatch( func processPolicyRulesBatch(
boltdb *bolt.DB, boltdb *bolt.DB,
batch []EventFollower, batch []EventTraveller,
queuedChan, skippedChan chan EventFollower, queuedChan, skippedChan chan EventTraveller,
) { ) {
eventIDs := make([]string, 0, len(batch)) eventIDs := make([]string, 0, len(batch))
for _, follower := range batch { for _, traveller := range batch {
eventIDs = append(eventIDs, follower.ID) eventIDs = append(eventIDs, traveller.ID)
} }
existsMap := BatchCheckEventsExist(boltdb, eventIDs) existsMap := BatchCheckEventsExist(boltdb, eventIDs)
for _, follower := range batch { for _, traveller := range batch {
if existsMap[follower.ID] { if existsMap[traveller.ID] {
skippedChan <- follower skippedChan <- traveller
} else { } else {
queuedChan <- follower queuedChan <- traveller
} }
} }
} }
func convertEventsToSubgraphs( func convertEventsToSubgraphs(
wg *sync.WaitGroup, expanders ExpanderPipeline, wg *sync.WaitGroup, expanders ExpanderPipeline,
inChan, convertedChan chan EventFollower, inChan, convertedChan chan EventTraveller,
) { ) {
defer wg.Done() defer wg.Done()
for follower := range inChan { for traveller := range inChan {
subgraph := EventToSubgraph(follower.Event, expanders) subgraph := EventToSubgraph(traveller.Event, expanders)
follower.Subgraph = subgraph traveller.Subgraph = subgraph
convertedChan <- follower convertedChan <- traveller
} }
close(convertedChan) close(convertedChan)
} }
@@ -232,14 +232,14 @@ func convertEventsToSubgraphs(
func writeEventsToDatabases( func writeEventsToDatabases(
wg *sync.WaitGroup, wg *sync.WaitGroup,
driver neo4j.Driver, boltdb *bolt.DB, driver neo4j.Driver, boltdb *bolt.DB,
inChan chan EventFollower, inChan chan EventTraveller,
resultChan chan WriteResult, resultChan chan WriteResult,
) { ) {
defer wg.Done() defer wg.Done()
var localWg sync.WaitGroup var localWg sync.WaitGroup
boltEventChan := make(chan EventFollower) boltEventChan := make(chan EventTraveller)
graphEventChan := make(chan EventFollower) graphEventChan := make(chan EventTraveller)
boltErrorChan := make(chan error) boltErrorChan := make(chan error)
graphResultChan := make(chan WriteResult) graphResultChan := make(chan WriteResult)
@@ -249,9 +249,9 @@ func writeEventsToDatabases(
go writeEventsToGraphDB(&localWg, driver, graphEventChan, boltErrorChan, graphResultChan) go writeEventsToGraphDB(&localWg, driver, graphEventChan, boltErrorChan, graphResultChan)
// Fan out events to both writers // Fan out events to both writers
for follower := range inChan { for traveller := range inChan {
boltEventChan <- follower boltEventChan <- traveller
graphEventChan <- follower graphEventChan <- traveller
} }
close(boltEventChan) close(boltEventChan)
close(graphEventChan) close(graphEventChan)
@@ -265,15 +265,15 @@ func writeEventsToDatabases(
func writeEventsToBoltDB( func writeEventsToBoltDB(
wg *sync.WaitGroup, wg *sync.WaitGroup,
boltdb *bolt.DB, boltdb *bolt.DB,
inChan chan EventFollower, inChan chan EventTraveller,
errorChan chan error, errorChan chan error,
) { ) {
defer wg.Done() defer wg.Done()
var events []EventBlob var events []EventBlob
for follower := range inChan { for traveller := range inChan {
events = append(events, events = append(events,
EventBlob{ID: follower.ID, JSON: follower.JSON}) EventBlob{ID: traveller.ID, JSON: traveller.JSON})
} }
err := BatchWriteEvents(boltdb, events) err := BatchWriteEvents(boltdb, events)
@@ -285,7 +285,7 @@ func writeEventsToBoltDB(
func writeEventsToGraphDB( func writeEventsToGraphDB(
wg *sync.WaitGroup, wg *sync.WaitGroup,
driver neo4j.Driver, driver neo4j.Driver,
inChan chan EventFollower, inChan chan EventTraveller,
boltErrorChan chan error, boltErrorChan chan error,
resultChan chan WriteResult, resultChan chan WriteResult,
) { ) {
@@ -293,11 +293,11 @@ func writeEventsToGraphDB(
matchKeys := NewSimpleMatchKeys() matchKeys := NewSimpleMatchKeys()
batch := NewBatchSubgraph(matchKeys) batch := NewBatchSubgraph(matchKeys)
for follower := range inChan { for traveller := range inChan {
for _, node := range follower.Subgraph.Nodes() { for _, node := range traveller.Subgraph.Nodes() {
batch.AddNode(node) batch.AddNode(node)
} }
for _, rel := range follower.Subgraph.Rels() { for _, rel := range traveller.Subgraph.Rels() {
batch.AddRel(rel) batch.AddRel(rel)
} }
} }
@@ -320,11 +320,11 @@ func writeEventsToGraphDB(
close(resultChan) close(resultChan)
} }
func collectEvents(wg *sync.WaitGroup, inChan chan EventFollower, resultChan chan []EventFollower) { func collectTravellers(wg *sync.WaitGroup, inChan chan EventTraveller, resultChan chan []EventTraveller) {
defer wg.Done() defer wg.Done()
var collected []EventFollower var collected []EventTraveller
for follower := range inChan { for traveller := range inChan {
collected = append(collected, follower) collected = append(collected, traveller)
} }
resultChan <- collected resultChan <- collected
close(resultChan) close(resultChan)

64
write_test.go
View File

@@ -20,28 +20,28 @@ func invalidEventJSON() string {
// Pipeline stage tests // Pipeline stage tests
func TestCreateEventFollowers(t *testing.T) { func TestCreateEventTravellers(t *testing.T) {
cases := []struct { cases := []struct {
name string name string
input []string input []string
expected []EventFollower expected []EventTraveller
}{ }{
{ {
name: "empty input", name: "empty input",
input: []string{}, input: []string{},
expected: []EventFollower{}, expected: []EventTraveller{},
}, },
{ {
name: "single json", name: "single json",
input: []string{"test1"}, input: []string{"test1"},
expected: []EventFollower{ expected: []EventTraveller{
{JSON: "test1"}, {JSON: "test1"},
}, },
}, },
{ {
name: "multiple jsons", name: "multiple jsons",
input: []string{"test1", "test2", "test3"}, input: []string{"test1", "test2", "test3"},
expected: []EventFollower{ expected: []EventTraveller{
{JSON: "test1"}, {JSON: "test1"},
{JSON: "test2"}, {JSON: "test2"},
{JSON: "test3"}, {JSON: "test3"},
@@ -53,10 +53,10 @@ func TestCreateEventFollowers(t *testing.T) {
t.Run(tc.name, func(t *testing.T) { t.Run(tc.name, func(t *testing.T) {
var wg sync.WaitGroup var wg sync.WaitGroup
jsonChan := make(chan string) jsonChan := make(chan string)
eventChan := make(chan EventFollower) eventChan := make(chan EventTraveller)
wg.Add(1) wg.Add(1)
go createEventFollowers(&wg, jsonChan, eventChan) go createEventTravellers(&wg, jsonChan, eventChan)
go func() { go func() {
for _, raw := range tc.input { for _, raw := range tc.input {
@@ -65,9 +65,9 @@ func TestCreateEventFollowers(t *testing.T) {
close(jsonChan) close(jsonChan)
}() }()
var result []EventFollower var result []EventTraveller
for follower := range eventChan { for traveller := range eventChan {
result = append(result, follower) result = append(result, traveller)
} }
wg.Wait() wg.Wait()
@@ -84,7 +84,7 @@ func TestCreateEventFollowers(t *testing.T) {
func TestParseEventJSON(t *testing.T) { func TestParseEventJSON(t *testing.T) {
cases := []struct { cases := []struct {
name string name string
input []EventFollower input []EventTraveller
wantParsed int wantParsed int
wantInvalid int wantInvalid int
checkParsedID bool checkParsedID bool
@@ -92,7 +92,7 @@ func TestParseEventJSON(t *testing.T) {
}{ }{
{ {
name: "valid event", name: "valid event",
input: []EventFollower{ input: []EventTraveller{
{JSON: validEventJSON("abc123", "pubkey1")}, {JSON: validEventJSON("abc123", "pubkey1")},
}, },
wantParsed: 1, wantParsed: 1,
@@ -102,7 +102,7 @@ func TestParseEventJSON(t *testing.T) {
}, },
{ {
name: "invalid json", name: "invalid json",
input: []EventFollower{ input: []EventTraveller{
{JSON: invalidEventJSON()}, {JSON: invalidEventJSON()},
}, },
wantParsed: 0, wantParsed: 0,
@@ -110,7 +110,7 @@ func TestParseEventJSON(t *testing.T) {
}, },
{ {
name: "mixed batch", name: "mixed batch",
input: []EventFollower{ input: []EventTraveller{
{JSON: validEventJSON("abc123", "pubkey1")}, {JSON: validEventJSON("abc123", "pubkey1")},
{JSON: invalidEventJSON()}, {JSON: invalidEventJSON()},
{JSON: validEventJSON("def456", "pubkey2")}, {JSON: validEventJSON("def456", "pubkey2")},
@@ -123,22 +123,22 @@ func TestParseEventJSON(t *testing.T) {
for _, tc := range cases { for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) { t.Run(tc.name, func(t *testing.T) {
var wg sync.WaitGroup var wg sync.WaitGroup
inChan := make(chan EventFollower) inChan := make(chan EventTraveller)
parsedChan := make(chan EventFollower) parsedChan := make(chan EventTraveller)
invalidChan := make(chan EventFollower) invalidChan := make(chan EventTraveller)
wg.Add(1) wg.Add(1)
go parseEventJSON(&wg, inChan, parsedChan, invalidChan) go parseEventJSON(&wg, inChan, parsedChan, invalidChan)
go func() { go func() {
for _, follower := range tc.input { for _, traveller := range tc.input {
inChan <- follower inChan <- traveller
} }
close(inChan) close(inChan)
}() }()
var parsed []EventFollower var parsed []EventTraveller
var invalid []EventFollower var invalid []EventTraveller
var collectWg sync.WaitGroup var collectWg sync.WaitGroup
collectWg.Add(2) collectWg.Add(2)
@@ -204,8 +204,8 @@ func TestConvertEventsToSubgraphs(t *testing.T) {
for _, tc := range cases { for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) { t.Run(tc.name, func(t *testing.T) {
var wg sync.WaitGroup var wg sync.WaitGroup
inChan := make(chan EventFollower) inChan := make(chan EventTraveller)
convertedChan := make(chan EventFollower) convertedChan := make(chan EventTraveller)
expanders := NewExpanderPipeline(DefaultExpanders()...) expanders := NewExpanderPipeline(DefaultExpanders()...)
@@ -213,11 +213,11 @@ func TestConvertEventsToSubgraphs(t *testing.T) {
go convertEventsToSubgraphs(&wg, expanders, inChan, convertedChan) go convertEventsToSubgraphs(&wg, expanders, inChan, convertedChan)
go func() { go func() {
inChan <- EventFollower{Event: tc.event} inChan <- EventTraveller{Event: tc.event}
close(inChan) close(inChan)
}() }()
var result EventFollower var result EventTraveller
for f := range convertedChan { for f := range convertedChan {
result = f result = f
} }
@@ -236,17 +236,17 @@ func TestConvertEventsToSubgraphs(t *testing.T) {
func TestCollectEvents(t *testing.T) { func TestCollectEvents(t *testing.T) {
cases := []struct { cases := []struct {
name string name string
input []EventFollower input []EventTraveller
expected int expected int
}{ }{
{ {
name: "empty channel", name: "empty channel",
input: []EventFollower{}, input: []EventTraveller{},
expected: 0, expected: 0,
}, },
{ {
name: "multiple followers", name: "multiple travellers",
input: []EventFollower{ input: []EventTraveller{
{ID: "id1"}, {ID: "id1"},
{ID: "id2"}, {ID: "id2"},
{ID: "id3"}, {ID: "id3"},
@@ -258,11 +258,11 @@ func TestCollectEvents(t *testing.T) {
for _, tc := range cases { for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) { t.Run(tc.name, func(t *testing.T) {
var wg sync.WaitGroup var wg sync.WaitGroup
inChan := make(chan EventFollower) inChan := make(chan EventTraveller)
resultChan := make(chan []EventFollower) resultChan := make(chan []EventTraveller)
wg.Add(1) wg.Add(1)
go collectEvents(&wg, inChan, resultChan) go collectTravellers(&wg, inChan, resultChan)
go func() { go func() {
for _, f := range tc.input { for _, f := range tc.input {