// Copyright (c) 2020-2024 Blockwatch Data Inc.
// Author: alex@blockwatch.cc

package tables

import (
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"strconv"
	"strings"
	"time"

	"blockwatch.cc/packdb/encoding/csv"
	"blockwatch.cc/packdb/pack"
	"blockwatch.cc/packdb/util"
	"blockwatch.cc/tzindex/etl/model"
	"blockwatch.cc/tzindex/rpc"
	"blockwatch.cc/tzindex/server"
)

var (
	// long -> short form
	supplySourceNames map[string]string
	// all aliases as list
	supplyAllAliases []string
)

func init() {
	fields, err := pack.Fields(&model.Supply{})
	if err != nil {
		log.Fatalf("supply field type error: %v\n", err)
	}
	supplySourceNames = fields.NameMapReverse()
	supplyAllAliases = fields.Aliases()
}

// configurable marshalling helper
type Supply struct {
	model.Supply
	verbose bool            // cond. marshal
	columns util.StringList // cond. cols & order when brief
	params  *rpc.Params     // blockchain amount conversion
}

func (s *Supply) MarshalJSON() ([]byte, error) {
	if s.verbose {
		return s.MarshalJSONVerbose()
	} else {
		return s.MarshalJSONBrief()
	}
}

func (s *Supply) MarshalJSONVerbose() ([]byte, error) {
	supply := struct {
		RowId               uint64    `json:"row_id"`
		Height              int64     `json:"height"`
		Cycle               int64     `json:"cycle"`
		Timestamp           time.Time `json:"time"`
		Total               float64   `json:"total"`
		Activated           float64   `json:"activated"`
		Unclaimed           float64   `json:"unclaimed"`
		Circulating         float64   `json:"circulating"`
		Liquid              float64   `json:"liquid"`
		Delegated           float64   `json:"delegated"`
		Staking             float64   `json:"staking"`
		Unstaking           float64   `json:"unstaking"`
		Shielded            float64   `json:"shielded"`
		ActiveStake         float64   `json:"active_stake"`
		ActiveDelegated     float64   `json:"active_delegated"`
		ActiveStaking       float64   `json:"active_staking"`
		InactiveDelegated   float64   `json:"inactive_delegated"`
		InactiveStaking     float64   `json:"inactive_staking"`
		Minted              float64   `json:"minted"`
		MintedBaking        float64   `json:"minted_baking"`
		MintedEndorsing     float64   `json:"minted_endorsing"`
		MintedSeeding       float64   `json:"minted_seeding"`
		MintedAirdrop       float64   `json:"minted_airdrop"`
		MintedSubsidy       float64   `json:"minted_subsidy"`
		Burned              float64   `json:"burned"`
		BurnedDoubleBaking  float64   `json:"burned_double_baking"`
		BurnedDoubleEndorse float64   `json:"burned_double_endorse"`
		BurnedOrigination   float64   `json:"burned_origination"`
		BurnedAllocation    float64   `json:"burned_allocation"`
		BurnedStorage       float64   `json:"burned_storage"`
		BurnedExplicit      float64   `json:"burned_explicit"`
		BurnedSeedMiss      float64   `json:"burned_seed_miss"`
		BurnedOffline       float64   `json:"burned_offline"`
		BurnedRollup        float64   `json:"burned_rollup"`
		Frozen              float64   `json:"frozen"`
		FrozenDeposits      float64   `json:"frozen_deposits"`
		FrozenRewards       float64   `json:"frozen_rewards"`
		FrozenFees          float64   `json:"frozen_fees"`
		FrozenBonds         float64   `json:"frozen_bonds"`
		FrozenStake         float64   `json:"frozen_stake"`
		FrozenBakerStake    float64   `json:"frozen_baker_stake"`
		FrozenStakerStake   float64   `json:"frozen_staker_stake"`
	}{
		RowId:               s.RowId,
		Height:              s.Height,
		Cycle:               s.Cycle,
		Timestamp:           s.Timestamp,
		Total:               s.params.ConvertValue(s.Total),
		Activated:           s.params.ConvertValue(s.Activated),
		Unclaimed:           s.params.ConvertValue(s.Unclaimed),
		Circulating:         s.params.ConvertValue(s.Circulating),
		Liquid:              s.params.ConvertValue(s.Liquid),
		Delegated:           s.params.ConvertValue(s.Delegated),
		Staking:             s.params.ConvertValue(s.Staking),
		Unstaking:           s.params.ConvertValue(s.Unstaking),
		Shielded:            s.params.ConvertValue(s.Shielded),
		ActiveStake:         s.params.ConvertValue(s.ActiveStake),
		ActiveDelegated:     s.params.ConvertValue(s.ActiveDelegated),
		ActiveStaking:       s.params.ConvertValue(s.ActiveStaking),
		InactiveDelegated:   s.params.ConvertValue(s.InactiveDelegated),
		InactiveStaking:     s.params.ConvertValue(s.InactiveStaking),
		Minted:              s.params.ConvertValue(s.Minted),
		MintedBaking:        s.params.ConvertValue(s.MintedBaking),
		MintedEndorsing:     s.params.ConvertValue(s.MintedEndorsing),
		MintedSeeding:       s.params.ConvertValue(s.MintedSeeding),
		MintedAirdrop:       s.params.ConvertValue(s.MintedAirdrop),
		MintedSubsidy:       s.params.ConvertValue(s.MintedSubsidy),
		Burned:              s.params.ConvertValue(s.Burned),
		BurnedDoubleBaking:  s.params.ConvertValue(s.BurnedDoubleBaking),
		BurnedDoubleEndorse: s.params.ConvertValue(s.BurnedDoubleEndorse),
		BurnedOrigination:   s.params.ConvertValue(s.BurnedOrigination),
		BurnedAllocation:    s.params.ConvertValue(s.BurnedAllocation),
		BurnedStorage:       s.params.ConvertValue(s.BurnedStorage),
		BurnedExplicit:      s.params.ConvertValue(s.BurnedExplicit),
		BurnedSeedMiss:      s.params.ConvertValue(s.BurnedSeedMiss),
		BurnedOffline:       s.params.ConvertValue(s.BurnedOffline),
		BurnedRollup:        s.params.ConvertValue(s.BurnedRollup),
		Frozen:              s.params.ConvertValue(s.Frozen),
		FrozenDeposits:      s.params.ConvertValue(s.FrozenDeposits),
		FrozenRewards:       s.params.ConvertValue(s.FrozenRewards),
		FrozenFees:          s.params.ConvertValue(s.FrozenFees),
		FrozenBonds:         s.params.ConvertValue(s.FrozenBonds),
		FrozenStake:         s.params.ConvertValue(s.FrozenStake),
		FrozenBakerStake:    s.params.ConvertValue(s.FrozenBakerStake),
		FrozenStakerStake:   s.params.ConvertValue(s.FrozenStakerStake),
	}
	return json.Marshal(supply)
}

func (s *Supply) MarshalJSONBrief() ([]byte, error) {
	dec := s.params.Decimals
	buf := make([]byte, 0, 2048)
	buf = append(buf, '[')
	for i, v := range s.columns {
		switch v {
		case "row_id":
			buf = strconv.AppendUint(buf, s.RowId, 10)
		case "height":
			buf = strconv.AppendInt(buf, s.Height, 10)
		case "cycle":
			buf = strconv.AppendInt(buf, s.Cycle, 10)
		case "time":
			buf = strconv.AppendInt(buf, util.UnixMilliNonZero(s.Timestamp), 10)
		case "total":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.Total), 'f', dec, 64)
		case "activated":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.Activated), 'f', dec, 64)
		case "unclaimed":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.Unclaimed), 'f', dec, 64)
		case "circulating":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.Circulating), 'f', dec, 64)
		case "liquid":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.Liquid), 'f', dec, 64)
		case "delegated":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.Delegated), 'f', dec, 64)
		case "staking":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.Staking), 'f', dec, 64)
		case "unstaking":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.Unstaking), 'f', dec, 64)
		case "shielded":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.Shielded), 'f', dec, 64)
		case "active_stake":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.ActiveStake), 'f', dec, 64)
		case "active_delegated":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.ActiveDelegated), 'f', dec, 64)
		case "active_staking":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.ActiveStaking), 'f', dec, 64)
		case "inactive_delegated":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.InactiveDelegated), 'f', dec, 64)
		case "inactive_staking":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.InactiveStaking), 'f', dec, 64)
		case "minted":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.Minted), 'f', dec, 64)
		case "minted_baking":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.MintedBaking), 'f', dec, 64)
		case "minted_endorsing":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.MintedEndorsing), 'f', dec, 64)
		case "minted_seeding":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.MintedSeeding), 'f', dec, 64)
		case "minted_airdrop":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.MintedAirdrop), 'f', dec, 64)
		case "minted_subsidy":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.MintedSubsidy), 'f', dec, 64)
		case "burned":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.Burned), 'f', dec, 64)
		case "burned_double_baking":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.BurnedDoubleBaking), 'f', dec, 64)
		case "burned_double_endorse":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.BurnedDoubleEndorse), 'f', dec, 64)
		case "burned_origination":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.BurnedOrigination), 'f', dec, 64)
		case "burned_allocation":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.BurnedAllocation), 'f', dec, 64)
		case "burned_storage":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.BurnedStorage), 'f', dec, 64)
		case "burned_explicit":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.BurnedExplicit), 'f', dec, 64)
		case "burned_seed_miss":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.BurnedSeedMiss), 'f', dec, 64)
		case "burned_offline":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.BurnedOffline), 'f', dec, 64)
		case "burned_rollup":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.BurnedRollup), 'f', dec, 64)
		case "frozen":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.Frozen), 'f', dec, 64)
		case "frozen_deposits":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.FrozenDeposits), 'f', dec, 64)
		case "frozen_rewards":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.FrozenRewards), 'f', dec, 64)
		case "frozen_fees":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.FrozenFees), 'f', dec, 64)
		case "frozen_bonds":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.FrozenBonds), 'f', dec, 64)
		case "frozen_stake":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.FrozenStake), 'f', dec, 64)
		case "frozen_baker_stake":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.FrozenBakerStake), 'f', dec, 64)
		case "frozen_staker_stake":
			buf = strconv.AppendFloat(buf, s.params.ConvertValue(s.FrozenStakerStake), 'f', dec, 64)
		default:
			continue
		}
		if i < len(s.columns)-1 {
			buf = append(buf, ',')
		}
	}
	buf = append(buf, ']')
	return buf, nil
}

func (s *Supply) MarshalCSV() ([]string, error) {
	dec := s.params.Decimals
	res := make([]string, len(s.columns))
	for i, v := range s.columns {
		switch v {
		case "row_id":
			res[i] = strconv.FormatUint(s.RowId, 10)
		case "height":
			res[i] = strconv.FormatInt(s.Height, 10)
		case "cycle":
			res[i] = strconv.FormatInt(s.Cycle, 10)
		case "time":
			res[i] = strconv.Quote(s.Timestamp.Format(time.RFC3339))
		case "total":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.Total), 'f', dec, 64)
		case "activated":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.Activated), 'f', dec, 64)
		case "unclaimed":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.Unclaimed), 'f', dec, 64)
		case "circulating":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.Circulating), 'f', dec, 64)
		case "liquid":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.Liquid), 'f', dec, 64)
		case "delegated":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.Delegated), 'f', dec, 64)
		case "staking":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.Staking), 'f', dec, 64)
		case "unstaking":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.Unstaking), 'f', dec, 64)
		case "shielded":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.Shielded), 'f', dec, 64)
		case "active_stake":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.ActiveStake), 'f', dec, 64)
		case "active_delegated":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.ActiveDelegated), 'f', dec, 64)
		case "active_staking":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.ActiveStaking), 'f', dec, 64)
		case "inactive_delegated":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.InactiveDelegated), 'f', dec, 64)
		case "inactive_staking":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.InactiveStaking), 'f', dec, 64)
		case "minted":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.Minted), 'f', dec, 64)
		case "minted_baking":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.MintedBaking), 'f', dec, 64)
		case "minted_endorsing":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.MintedEndorsing), 'f', dec, 64)
		case "minted_seeding":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.MintedSeeding), 'f', dec, 64)
		case "minted_airdrop":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.MintedAirdrop), 'f', dec, 64)
		case "minted_subsidy":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.MintedSubsidy), 'f', dec, 64)
		case "burned":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.Burned), 'f', dec, 64)
		case "burned_double_baking":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.BurnedDoubleBaking), 'f', dec, 64)
		case "burned_double_endorse":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.BurnedDoubleEndorse), 'f', dec, 64)
		case "burned_origination":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.BurnedOrigination), 'f', dec, 64)
		case "burned_allocation":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.BurnedAllocation), 'f', dec, 64)
		case "burned_storage":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.BurnedStorage), 'f', dec, 64)
		case "burned_explicit":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.BurnedExplicit), 'f', dec, 64)
		case "burned_seed_miss":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.BurnedSeedMiss), 'f', dec, 64)
		case "burned_offline":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.BurnedOffline), 'f', dec, 64)
		case "burned_rollup":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.BurnedRollup), 'f', dec, 64)
		case "frozen":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.Frozen), 'f', dec, 64)
		case "frozen_deposits":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.FrozenDeposits), 'f', dec, 64)
		case "frozen_rewards":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.FrozenRewards), 'f', dec, 64)
		case "frozen_fees":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.FrozenFees), 'f', dec, 64)
		case "frozen_bonds":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.FrozenBonds), 'f', dec, 64)
		case "frozen_stake":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.FrozenStake), 'f', dec, 64)
		case "frozen_baker_stake":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.FrozenBakerStake), 'f', dec, 64)
		case "frozen_staker_stake":
			res[i] = strconv.FormatFloat(s.params.ConvertValue(s.FrozenStakerStake), 'f', dec, 64)
		default:
			continue
		}
	}
	return res, nil
}

func StreamSupplyTable(ctx *server.Context, args *TableRequest) (interface{}, int) {
	// use chain params at current height
	params := ctx.Params

	// access table
	table, err := ctx.Indexer.Table(args.Table)
	if err != nil {
		panic(server.ENotFound(server.EC_RESOURCE_NOTFOUND, fmt.Sprintf("cannot access table '%s'", args.Table), err))
	}

	// translate long column names to short names used in pack tables
	var srcNames []string
	if len(args.Columns) > 0 {
		// resolve short column names
		srcNames = make([]string, 0, len(args.Columns))
		for _, v := range args.Columns {
			n, ok := supplySourceNames[v]
			if !ok {
				panic(server.EBadRequest(server.EC_PARAM_INVALID, fmt.Sprintf("unknown column '%s'", v), nil))
			}
			if n != "-" {
				srcNames = append(srcNames, n)
			}
		}
	} else {
		// use all table columns in order and reverse lookup their long names
		srcNames = table.Fields().Names()
		args.Columns = supplyAllAliases
	}

	// build table query
	q := pack.NewQuery(ctx.RequestID).
		WithTable(table).
		WithFields(srcNames...).
		WithLimit(int(args.Limit)).
		WithOrder(args.Order)

	// build dynamic filter conditions from query (will panic on error)
	for key, val := range ctx.Request.URL.Query() {
		keys := strings.Split(key, ".")
		prefix := keys[0]
		mode := pack.FilterModeEqual
		if len(keys) > 1 {
			mode = pack.ParseFilterMode(keys[1])
			if !mode.IsValid() {
				panic(server.EBadRequest(server.EC_PARAM_INVALID, fmt.Sprintf("invalid filter mode '%s'", keys[1]), nil))
			}
		}
		switch prefix {
		case "columns", "limit", "order", "verbose", "filename":
			// skip these fields
		case "cursor":
			// add row id condition: id > cursor (new cursor == last row id)
			id, err := strconv.ParseUint(val[0], 10, 64)
			if err != nil {
				panic(server.EBadRequest(server.EC_PARAM_INVALID, fmt.Sprintf("invalid cursor value '%s'", val), err))
			}
			cursorMode := pack.FilterModeGt
			if args.Order == pack.OrderDesc {
				cursorMode = pack.FilterModeLt
			}
			q = q.And("I", cursorMode, id)
		default:
			// translate long column name used in query to short column name used in packs
			if short, ok := supplySourceNames[prefix]; !ok {
				panic(server.EBadRequest(server.EC_PARAM_INVALID, fmt.Sprintf("unknown column '%s'", prefix), nil))
			} else {
				key = strings.Replace(key, prefix, short, 1)
			}

			// the same field name may appear multiple times, in which case conditions
			// are combined like any other condition with logical AND
			for _, v := range val {

				switch prefix {
				case "height", "time":
					// need no conversion
				default:
					fvals := make([]string, 0)
					for _, vv := range strings.Split(v, ",") {
						fval, err := strconv.ParseFloat(vv, 64)
						if err != nil {
							panic(server.EBadRequest(server.EC_PARAM_INVALID, fmt.Sprintf("invalid %s filter value '%s'", key, vv), err))
						}
						fvals = append(fvals, strconv.FormatInt(params.ConvertAmount(fval), 10))
					}
					v = strings.Join(fvals, ",")
				}

				if cond, err := pack.ParseCondition(key, v, table.Fields()); err != nil {
					panic(server.EBadRequest(server.EC_PARAM_INVALID, fmt.Sprintf("invalid %s filter value '%s'", key, v), err))
				} else {
					q = q.AndCondition(cond)
				}
			}
		}
	}

	var (
		count  int
		lastId uint64
	)

	// start := time.Now()
	// ctx.Log.Tracef("Streaming max %d rows from %s", args.Limit, args.Table)
	// defer func() {
	// 	ctx.Log.Tracef("Streamed %d rows in %s", count, time.Since(start))
	// }()

	// prepare return type marshalling
	supply := &Supply{
		verbose: args.Verbose,
		columns: args.Columns,
		params:  params,
	}

	// prepare response stream
	ctx.StreamResponseHeaders(http.StatusOK, mimetypes[args.Format])

	switch args.Format {
	case "json":
		enc := json.NewEncoder(ctx.ResponseWriter)
		enc.SetIndent("", "")
		enc.SetEscapeHTML(false)

		// open JSON array
		_, _ = io.WriteString(ctx.ResponseWriter, "[")
		// close JSON array on panic
		defer func() {
			if e := recover(); e != nil {
				_, _ = io.WriteString(ctx.ResponseWriter, "]")
				panic(e)
			}
		}()

		// run query and stream results
		var needComma bool
		err = table.Stream(ctx, q, func(r pack.Row) error {
			if needComma {
				_, _ = io.WriteString(ctx.ResponseWriter, ",")
			} else {
				needComma = true
			}
			if err := r.Decode(supply); err != nil {
				return err
			}
			if err := enc.Encode(supply); err != nil {
				return err
			}
			count++
			lastId = supply.RowId
			if args.Limit > 0 && count == int(args.Limit) {
				return io.EOF
			}
			return nil
		})
		// close JSON bracket
		_, _ = io.WriteString(ctx.ResponseWriter, "]")
		// ctx.Log.Tracef("JSON encoded %d rows", count)

	case "csv":
		enc := csv.NewEncoder(ctx.ResponseWriter)
		// use custom header columns and order
		if len(args.Columns) > 0 {
			err = enc.EncodeHeader(args.Columns, nil)
		}
		if err == nil {
			// run query and stream results
			err = table.Stream(ctx, q, func(r pack.Row) error {
				if err := r.Decode(supply); err != nil {
					return err
				}
				if err := enc.EncodeRecord(supply); err != nil {
					return err
				}
				count++
				lastId = supply.RowId
				if args.Limit > 0 && count == int(args.Limit) {
					return io.EOF
				}
				return nil
			})
		}
		// ctx.Log.Tracef("CSV Encoded %d rows", count)
	}

	// without new records, cursor remains the same as input (may be empty)
	cursor := args.Cursor
	if lastId > 0 {
		cursor = strconv.FormatUint(lastId, 10)
	}

	// write error (except EOF), cursor and count as http trailer
	ctx.StreamTrailer(cursor, count, err)

	// streaming return
	return nil, -1
}
