aljaz/crawler_v2
1
0
Fork 0
mirror of https://github.com/aljazceru/crawler_v2.git synced 2025-12-17 07:24:21 +01:00
Code Issues Packages Projects Releases Wiki Activity

moved back to a single db writer

Browse Source
This commit is contained in:
pippellia-btc
2025-06-17 12:02:59 +02:00
parent 94d52f7a1c
commit 978494cb50
2 changed files with 27 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
pkg/config/config.go
View File

@@ -200,12 +200,6 @@ func Load() (*Config, error) {
if err != nil { if err != nil {
return nil, fmt.Errorf("error parsing %v: %v", keyVal, err) return nil, fmt.Errorf("error parsing %v: %v", keyVal, err)
} }
case "ENGINE_ARCHIVERS":
config.Engine.Archivers, err = strconv.Atoi(val)
if err != nil {
return nil, fmt.Errorf("error parsing %v: %v", keyVal, err)
}
} }
} }

55
pkg/pipe/engine.go
View File

@@ -24,9 +24,6 @@ type EngineConfig struct {
// GraphBuilder params // GraphBuilder params
BuilderCapacity int BuilderCapacity int
CacheCapacity int CacheCapacity int
// Archiver params
Archivers int
} }
func NewEngineConfig() EngineConfig { func NewEngineConfig() EngineConfig {
@@ -34,7 +31,6 @@ func NewEngineConfig() EngineConfig {
PrintEvery: 5000, PrintEvery: 5000,
BuilderCapacity: 1000, BuilderCapacity: 1000,
CacheCapacity: 100_000, CacheCapacity: 100_000,
Archivers: 4,
} }
} }
@@ -56,8 +52,10 @@ func Engine(
graphEvents := make(chan *nostr.Event, config.BuilderCapacity) graphEvents := make(chan *nostr.Event, config.BuilderCapacity)
defer close(graphEvents) defer close(graphEvents)
go GraphBuilder(ctx, config, db, graphEvents)
log.Println("Engine: ready to process events") log.Println("Engine: ready to process events")
defer log.Println("Engine: shutting down...")
go GraphBuilder(ctx, config, db, graphEvents)
Archiver(ctx, config, store, events, func(e *nostr.Event) error { Archiver(ctx, config, store, events, func(e *nostr.Event) error {
if e.Kind == nostr.KindFollowList { if e.Kind == nostr.KindFollowList {
@@ -69,8 +67,6 @@ func Engine(
} }
return nil return nil
}) })
log.Println("Engine: shutting down...")
} }
// Archiver stores events in the event store. // Archiver stores events in the event store.
@@ -81,7 +77,6 @@ func Archiver(
events chan *nostr.Event, events chan *nostr.Event,
onReplace func(*nostr.Event) error) { onReplace func(*nostr.Event) error) {
sem := make(chan struct{}, config.Archivers)
var processed int var processed int
for { for {
@@ -94,40 +89,16 @@ func Archiver(
return return
} }
sem <- struct{}{} err := func() error {
go func() { opctx, cancel := context.WithTimeout(ctx, 5*time.Second)
err := archive(ctx, store, event, onReplace)
<-sem
if err != nil {
log.Printf("Archiver: event ID %s, kind %d by %s: %v", event.ID, event.Kind, event.PubKey, err)
}
}()
processed++
if processed%config.PrintEvery == 0 {
log.Printf("Archiver: processed %d events", processed)
}
}
}
}
// Archive an event based on its kind.
func archive(
ctx context.Context,
store *eventstore.Store,
event *nostr.Event,
onReplace func(*nostr.Event) error) error {
ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
defer cancel() defer cancel()
switch { switch {
case nostr.IsRegularKind(event.Kind): case nostr.IsRegularKind(event.Kind):
return store.Save(ctx, event) return store.Save(opctx, event)
case nostr.IsReplaceableKind(event.Kind): case nostr.IsReplaceableKind(event.Kind):
replaced, err := store.Replace(ctx, event) replaced, err := store.Replace(opctx, event)
if err != nil { if err != nil {
return err return err
} }
@@ -140,6 +111,18 @@ func archive(
default: default:
return nil return nil
} }
}()
if err != nil {
log.Printf("Archiver: event ID %s, kind %d by %s: %v", event.ID, event.Kind, event.PubKey, err)
}
processed++
if processed%config.PrintEvery == 0 {
log.Printf("Archiver: processed %d events", processed)
}
}
}
} }
// GraphBuilder consumes events to update the graph and random walks. // GraphBuilder consumes events to update the graph and random walks.