package main

import (
	"bytes"
	"fmt"
	"io"
)

type CMArgs struct {
	eol  []byte
	eow  []byte
	wrap []byte
	pk   string
}

type CSVMerge struct {
	arg    CMArgs
	output io.WriteCloser
	fps    []*CMFile
	//
	lines   []Line
	lineCnt int64
	//
	totolLine int64
}

func (c *CSVMerge) Clean() {
	for _, fp := range c.fps {
		if fp != nil {
			fp.Close()
		}
	}
	if c.output != nil {
		c.output.Close()
	}
}

func (c *CSVMerge) Merge() error {
	wr := c.output
	c.lines = make([]Line, len(c.fps))

	for {
		line, err, finish := c.mergeOneLine()
		if err != nil {
			if err != io.EOF{
				LogErr("mergeOneLine err:%s", err.Error())
				return err
			}
		}
		c.lineCnt++
		wr.Write(line.bytes())
		if finish {
			break
		} else {
			wr.Write(c.arg.eol)
		}
	}
	LogInfo("process %d lines data.", c.lineCnt)
	return nil
}

func (c *CSVMerge) mergeOneLine() (Line, error, bool) {
	finish := false
	// 第一列是主键(pk).
	var err error
	for i := 0; i < len(c.fps); i++ {
		c.lines[i], err = c.fps[i].readLine()
		if err != nil {
			if err != io.EOF {
				return Line{}, err, finish
			}
			// LogInfo("finish")
			finish = true
		}
	}
	var line Line
	line.write(c.lines[0].bytes())
	pk0, _, err := c.lines[0].splitPK(c.arg.eow)
	if err != nil {
		// LogWarn("splitPK_0 err:%v", err)
		return Line{}, err, finish
	}
	for i := 1; i < len(c.lines); i++ {
		pk, sub, err := c.lines[i].splitPK(c.arg.eow)
		if err != nil {
			// LogWarn("splitPK_%d err:%v", i, err)
			return Line{}, err, finish
		}
		if !bytes.Equal(pk0, pk) {
			// LogWarn("splitPK_%d Equal err", i)
			return Line{}, fmt.Errorf("pk was not equal: %s_%d.%s != %s_%d.%s",
				c.lines[0].fname, c.lines[0].lineIdx, pk0,
				c.lines[i].fname, c.lines[i].lineIdx, pk), finish
		}
		line.write(c.arg.eow)
		line.write(sub)
	}
	// LogInfo("finish=%v", finish)
	return line, nil, finish
}