···

       
1
       
1
       
-
       
package main

     

       
2
       
2
       
-
       


     

       
3
       
3
       
-
       
import (

     

       
4
       
4
       
-
       
	"context"

     

       
5
       
5
       
-
       
	"flag"

     

       
6
       
6
       
-
       
	"fmt"

     

       
7
       
7
       
-
       
	"strings"

     

       
8
       
8
       
-
       
	"time"

     

       
9
       
9
       
-
       


     

       
10
       
10
       
-
       
	"tangled.sh/tangled.sh/core/knotclient"

     

       
11
       
11
       
-
       
)

     

       
12
       
12
       
-
       


     

       
13
       
13
       
-
       
func main() {

     

       
14
       
14
       
-
       
	knots := flag.String("knots", "", "list of knots to connect to")

     

       
15
       
15
       
-
       
	retryFlag := flag.Duration("retry", 1*time.Minute, "retry interval")

     

       
16
       
16
       
-
       
	maxRetryFlag := flag.Duration("max-retry", 30*time.Minute, "max retry interval")

     

       
17
       
17
       
-
       
	workerCount := flag.Int("workers", 10, "goroutine pool size")

     

       
18
       
18
       
-
       


     

       
19
       
19
       
-
       
	flag.Parse()

     

       
20
       
20
       
-
       


     

       
21
       
21
       
-
       
	if *knots == "" {

     

       
22
       
22
       
-
       
		fmt.Println("error: -knots is required")

     

       
23
       
23
       
-
       
		flag.Usage()

     

       
24
       
24
       
-
       
		return

     

       
25
       
25
       
-
       
	}

     

       
26
       
26
       
-
       


     

       
27
       
27
       
-
       
	ccfg := knotclient.ConsumerConfig{

     

       
28
       
28
       
-
       
		ProcessFunc:      processEvent,

     

       
29
       
29
       
-
       
		RetryInterval:    *retryFlag,

     

       
30
       
30
       
-
       
		MaxRetryInterval: *maxRetryFlag,

     

       
31
       
31
       
-
       
		WorkerCount:      *workerCount,

     

       
32
       
32
       
-
       
		Dev:              true,

     

       
33
       
33
       
-
       
	}

     

       
34
       
34
       
-
       
	for k := range strings.SplitSeq(*knots, ",") {

     

       
35
       
35
       
-
       
		ccfg.AddEventSource(knotclient.NewEventSource(k))

     

       
36
       
36
       
-
       
	}

     

       
37
       
37
       
-
       


     

       
38
       
38
       
-
       
	consumer := knotclient.NewEventConsumer(ccfg)

     

       
39
       
39
       
-
       


     

       
40
       
40
       
-
       
	ctx, cancel := context.WithCancel(context.Background())

     

       
41
       
41
       
-
       
	consumer.Start(ctx)

     

       
42
       
42
       
-
       
	time.Sleep(1 * time.Hour)

     

       
43
       
43
       
-
       
	cancel()

     

       
44
       
44
       
-
       
	consumer.Stop()

     

       
45
       
45
       
-
       
}

     

       
46
       
46
       
-
       


     

       
47
       
47
       
-
       
func processEvent(_ context.Context, source knotclient.EventSource, msg knotclient.Message) error {

     

       
48
       
48
       
-
       
	fmt.Printf("From %s (%s, %s): %s\n", source.Knot, msg.Rkey, msg.Nsid, string(msg.EventJson))

     

       
49
       
49
       
-
       
	return nil

     

       
50
       
50
       
-
       
}

     

···

       
37
       
37
       
 
       
		dateStr := day.Format("2006-01-02")

     

       
38
       
38
       
 
       
		_, err := stmt.Exec(did, dateStr, count)

     

       
39
       
39
       
 
       
		if err != nil {

     

       
40
       
40
       
-
       
			log.Println("Failed to insert for date %s: %v", dateStr, err)

     

       
40
       
40
       
+
       
			log.Printf("Failed to insert for date %s: %v", dateStr, err)

     

       
41
       
41
       
 
       
		}

     

       
42
       
42
       
 
       
	}

     

       
43
       
43
       
 
       


     

···

       
158
       
158
       
 
       
		fmt.Sprintf("--count"),

     

       
159
       
159
       
 
       
	)

     

       
160
       
160
       
 
       
	if err != nil {

     

       
161
       
161
       
-
       
		return 0, fmt.Errorf("failed to run rev-list", err)

     

       
161
       
161
       
+
       
		return 0, fmt.Errorf("failed to run rev-list: %w", err)

     

       
162
       
162
       
 
       
	}

     

       
163
       
163
       
 
       


     

       
164
       
164
       
 
       
	count, err := strconv.Atoi(strings.TrimSpace(string(output)))