package main

import (
	"bufio"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"log"
	"os"
	"sort"
	"strings"
	"sync"

	"code.justin.tv/rhys/nursery/pgparse"
)

type txid uint64
type spanid string
type ns int64

type record struct {
	Kind   recordType `json:"K"`
	Time   ns         `json:"T"`
	Server string     `json:"S"`
	Host   string     `json:"M"`
	Pid    int        `json:"P"`

	Tx   txid              `json:"I"`
	Span spanid            `json:"R"`
	Data map[string]string `json:"D"`
}

type railsTx struct {
	MainThumb string
	Tables2   [][]string
}

type recordType int

const (
	rtConnect   recordType = 1
	rtSpanStart recordType = 3
	rtSpanEnd   recordType = 4
	rtCallStart recordType = 5
	rtCallEnd   recordType = 6
)

func main() {
	var wg sync.WaitGroup

	jsonFile := "/dev/stdin"
	flag.StringVar(&jsonFile, "json", jsonFile, "File containing json-formatted trace events")
	flag.Parse()

	log.SetFlags(log.Ldate | log.Ltime | log.Lshortfile)

	file, err := os.Open(jsonFile)
	if err != nil {
		log.Fatalf("open err=%q", err)
	}

	defer func(c io.Closer) {
		cerr := c.Close()
		if cerr != nil {
			log.Fatalf("close err=%q", cerr)
		}
	}(file)

	ch := make(chan record, 100)
	inShards := make([]chan<- record, 0, 1<<4)
	records := make(chan map[txid]railsTx, cap(inShards))
	for i := 0; i < cap(inShards); i++ {
		ch := make(chan record, 100)
		inShards = append(inShards, ch)
		wg.Add(1)
		go func(ch <-chan record) {
			defer wg.Done()
			m := make(map[txid]railsTx)
			defer func() {
				records <- m
			}()
			var total, skipped int
			defer func() {
				log.Printf("total=%d skipped=%d", total, skipped)
			}()
			for i := 0; ; i++ {
				v, ok := <-ch
				if !ok {
					break
				}

				if skip := map[int]bool{
					4749:  true,
					11673: true,
					12627: true,
					15909: true,
					19883: true,
					28671: true,
					27040: true,
				}[i]; skip {
					// continue
				}

				var dirty bool
				tx := m[v.Tx]
				if v.Server == "rails" && v.Span == ".0" {
					tx.MainThumb = v.Data["thumb"]
					dirty = true
				}
				if tx.MainThumb == "" {
					// We're relying here on in-order processing, trying to
					// save work (and errors) when non-rails transactions
					// include sql (or weird python-interpolation sql).
					continue
				}
				if sql, ok := v.Data["sql"]; ok {
					if !strings.HasPrefix(strings.ToUpper(sql), "SELECT ") {
						continue
					}
					// if strings.HasPrefix(strings.ToUpper(sql), "COMMIT ") {
					// 	continue
					// }
					total++
					stmt, err := pgparse.Parse(sql)
					if err != nil {
						skipped++
						continue
						// log.Fatalf("pgparse i=%d err=%q sql=%s", i, err, sql)
					}
					tr := newStatementTracker()
					tr.dumpStmt(stmt)
					if err := tr.Err(); err != nil {
						log.Fatalf("dumpstmt i=%d err=%q sql=%s", i, err, sql)
					}
					tables := tr.Tables()
					tx.Tables2 = append(tx.Tables2, tables)
					dirty = true
				}
				if dirty {
					m[v.Tx] = tx
				}
			}
		}(ch)
	}

	go read(inShards, ch)

	log.Printf("reading")
	err = scan(ch, file)
	if err != nil {
		log.Fatalf("err=%q", err)
	}
	close(ch)
	wg.Wait()
	close(records)
	log.Printf("done")

	tablesByRoute := make(map[string]map[string]struct{})

	for rec := range records {
		for _, tx := range rec {
			route := tx.MainThumb
			tables, ok := tablesByRoute[route]
			if !ok {
				tables = make(map[string]struct{})
				tablesByRoute[route] = tables
			}
			for _, list := range tx.Tables2 {
				for _, table := range list {
					tables[table] = struct{}{}
				}
			}
		}
	}

	var routes []string
	for route := range tablesByRoute {
		routes = append(routes, route)
	}
	sort.Strings(routes)
	for _, route := range routes {
		tables := tablesByRoute[route]
		list := make([]string, 0, len(tables))
		for table := range tables {
			list = append(list, table)
		}
		sort.Strings(list)
		log.Printf("route=%q tables=%q", route, list)

		// output for golang.org/x/tools/cmd/digraph
		quoted := make([]string, 0, len(list))
		for _, table := range list {
			quoted = append(quoted, fmt.Sprintf("%q", "table "+table))
		}
		log.Printf("%q %s", "route "+route, strings.Join(quoted, " "))
	}
}

func scan(ch chan<- record, r io.Reader) error {
	sc := bufio.NewScanner(r)
	for i := 0; sc.Scan(); i++ {
		var v record

		err := json.Unmarshal(sc.Bytes(), &v)
		if err != nil {
			return err
		}

		ch <- v
	}

	err := sc.Err()
	if err != nil {
		return err
	}

	return nil
}

func read(out []chan<- record, in <-chan record) {
	defer func() {
		for _, ch := range out {
			close(ch)
		}
	}()
	for v := range in {
		out[int(v.Tx%txid(len(out)))] <- v
	}
}
