Version b

b.go

package main

import (
	"customerio/core/customers"
	"customerio/core/segments"
	"customerio/core/spaces"
	"flag"
	"log"
	"strconv"
	"time"

	"github.com/FoundationDB/fdb-go/fdb"
	"github.com/customerio/farm"
)

var id = flag.Int("env", 0, "environment to check")

func main() {
	flag.Parse()

	var count int
	var missing int
	var blank int
	env := segments.GetEnv(*id)

	start := time.Now()

	check := func(internals []string) {
		spaces.LowPriorityTransact(spaces.Db(), func(tr fdb.Transaction) {
			earliest := make(map[string]*customers.AttributeCommand)

			for _, internal := range internals {
				earliest[internal] = customers.NewAttributeCommand(env.Id, internal, "_earliest_event")
				earliest[internal].Prefetch(tr)
			}

			for _, internal := range internals {
				count += 1

				attr := earliest[internal].Commit(tr)

				if attr == "" {
					blank += 1
				}

				if ts, err := strconv.Atoi(attr); err != nil || ts == 0 {
					missing += 1
				}

				if count%1000 == 0 {
					log.Println(count, missing, blank)
				}
			}
		})
	}

	runner := farm.Run(
		20,
		func(in chan<- interface{}) error {
			batch := make([]string, 0, 10)

			customers.GetEnv(env.Id).EachInternal(0, "", func(internal string) bool {
				batch = append(batch, internal)

				if len(batch) >= 10 {
					in <- batch
					batch = make([]string, 0, 10)
				}

				return true
			})

			in <- batch

			return nil
		},
		func(in interface{}) (interface{}, error) {
			check(in.([]string))
			return nil, nil
		},
	)

	for _ = range runner.Results {
	}

	if err := <-runner.Errors; err != nil {
		panic(err)
	}

	log.Println("checked in", time.Since(start), count, missing, blank)
}