package main

import (
	"encoding/binary"
	"flag"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"os"
	"strconv"
	"strings"
	"sync"

	"code.justin.tv/release/trace/api"
	"code.justin.tv/release/trace/scanproto"
	"github.com/golang/protobuf/proto"
	"github.com/syndtr/goleveldb/leveldb"
	"github.com/syndtr/goleveldb/leveldb/opt"
)

const (
	dbShards         = 16
	unmarshalWorkers = 64
)

var (
	input = flag.String("input", "trace.data", "Input filename containing api.TransactionSet")
)

func main() {
	flag.Parse()

	var r io.Reader = strings.NewReader("")
	var c io.Closer = ioutil.NopCloser(nil)
	if *input != "" {
		f, err := os.Open(*input)
		if err != nil {
			log.Fatalf("file open err=%q", err)
		}
		c = f
		r = f
	}
	defer func() {
		err := c.Close()
		if err != nil {
			log.Printf("file close err=%q", err)
		}
	}()

	var dbs []*leveldb.DB
	for i := 0; i < dbShards; i++ {
		db, err := leveldb.OpenFile(fmt.Sprintf("db/shard-%d", i), &opt.Options{
			ErrorIfExist: true,

			// Compression: opt.NoCompression,

			BlockCacheCapacity:  128 << 20,
			BlockSize:           64 << 10,
			CompactionTableSize: 32 << 20,
			WriteBuffer:         64 << 20,
		})
		if err != nil {
			log.Fatalf("leveldb open err=%q", err)
		}
		defer func() {
			err := db.Close()
			if err != nil {
				log.Printf("leveldb close err=%q", err)
			}
		}()
		dbs = append(dbs, db)
	}

	sets := make(chan []byte)
	txs := make(chan *api.Transaction)

	var wg sync.WaitGroup
	for i := 0; i < unmarshalWorkers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for buf := range sets {
				var set api.TransactionSet
				err := proto.Unmarshal(buf, &set)
				if err != nil {
					continue
				}
				for _, tx := range set.Transaction {
					txs <- tx
				}
			}
		}()
	}

	go func() {
		wg.Wait()
		close(txs)
	}()

	go func() {
		// This scanner is nominally for EventSet inputs, but it will work
		// correctly for TransactionSet inputs as well since the two message types
		// share structure.
		sc := scanproto.NewEventSetScanner(r)
		sc.Buffer(nil, 1<<20)

		for sc.Scan() {
			buf := sc.Bytes()
			sets <- append([]byte(nil), buf...)
		}

		err := sc.Err()
		if err != nil {
			log.Fatalf("scan err=%q", err)
		}

		close(sets)
	}()

	var dbwg sync.WaitGroup
	for _, db := range dbs {
		dbwg.Add(1)
		go func(db *leveldb.DB) {
			defer dbwg.Done()

			batch := new(leveldb.Batch)
			var key []byte

			for tx := range txs {
				var txid [16]byte
				for i := 0; i < 2; i++ {
					if len(tx.TransactionId) < i {
						break
					}
					binary.LittleEndian.PutUint64(txid[i*8:i*8+8], tx.TransactionId[i])
				}

				var f func(*api.Call)
				f = func(c *api.Call) {
					p := callProc(c)
					for _, c := range c.GetSubcalls() {
						f(c)
					}
					key = key[:0]
					key = append(key, "index "...)
					key = strconv.AppendQuote(key, p.Svcname)
					key = append(key, " "...)
					key = strconv.AppendQuote(key, p.Host)
					key = append(key, " "...)
					key = strconv.AppendInt(key, int64(p.Pid), 10)
					key = append(key, " "...)
					key = append(key, txid[:]...)
					batch.Put(key, nil)
				}
				f(tx.GetRoot())

				val, err := proto.Marshal(tx)
				if err != nil {
					continue
				}

				key = key[:0]
				key = append(key, "txid"...)
				key = append(key, txid[:]...)
				batch.Put(key, val)

				if batch.Len() >= 10000 {
					db.Write(batch, nil)
					batch.Reset()
				}
			}

			db.Write(batch, nil)
		}(db)
	}

	dbwg.Wait()
}

func callProc(c *api.Call) proc {
	svc := c.GetSvc()
	if svc == nil {
		return proc{}
	}
	return proc{
		Svcname: svc.Name,
		Host:    svc.Host,
		Pid:     svc.Pid,
	}
}

type proc struct {
	Svcname string
	Host    string
	Pid     int32
}

type procs []proc

func (p procs) Len() int      { return len(p) }
func (p procs) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
func (p procs) Less(i, j int) bool {
	pi, pj := p[i], p[j]
	if pi.Svcname != pj.Svcname {
		return pi.Svcname < pj.Svcname
	}
	if pi.Host != pj.Host {
		return pi.Host < pj.Host
	}
	if pi.Pid != pj.Pid {
		return pi.Pid < pj.Pid
	}
	return false
}
