package logstorage

import (
	"fmt"
	"sync"

	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"

	"github.com/VictoriaMetrics/VictoriaLogs/lib/prefixfilter"
)

// filterPatternMatch filters field entries by the given pattern, which could be generated by collapse_nums pipe.
//
// See https://docs.victoriametrics.com/victorialogs/logsql/#collapse_nums-pipe
type filterPatternMatch struct {
	fieldName string

	funcName string
	pm       *patternMatcher

	tokensOnce   sync.Once
	tokens       []string
	tokensHashes []uint64
}

func (fp *filterPatternMatch) String() string {
	return fmt.Sprintf("%s%s(%s)", quoteFieldNameIfNeeded(fp.fieldName), fp.funcName, quoteTokenIfNeeded(fp.pm.String()))
}

func (fp *filterPatternMatch) updateNeededFields(pf *prefixfilter.Filter) {
	pf.AddAllowFilter(fp.fieldName)
}

func (fp *filterPatternMatch) getTokens() []string {
	fp.tokensOnce.Do(fp.initTokens)
	return fp.tokens
}

func (fp *filterPatternMatch) getTokensHashes() []uint64 {
	fp.tokensOnce.Do(fp.initTokens)
	return fp.tokensHashes
}

func (fp *filterPatternMatch) initTokens() {
	var a []string

	separators := fp.pm.separators

	switch fp.pm.pmo {
	case patternMatcherOptionAny:
		if len(separators) == 1 {
			sep := skipFirstToken(separators[0])
			sep = skipLastToken(sep)
			a = append(a, sep)
			separators = nil
			break
		}

		sep := skipFirstToken(separators[0])
		if sep != "" && isSpecialNumStart(sep[len(sep)-1]) {
			sep = skipLastToken(sep)
		}
		a = append(a, sep)
		separators = separators[1:]

		if len(separators) > 0 {
			sep := skipLastToken(separators[len(separators)-1])
			if sep != "" && isSpecialNumEnd(sep[0]) {
				sep = skipFirstToken(sep)
			}
			a = append(a, sep)
			separators = separators[:len(separators)-1]
		}
	case patternMatcherOptionFull:
		if len(separators) == 1 {
			a = append(a, separators[0])
			separators = nil
			break
		}

		sep := separators[0]
		if sep != "" && isSpecialNumStart(sep[len(sep)-1]) {
			sep = skipLastToken(sep)
		}
		a = append(a, sep)
		separators = separators[1:]

		if len(separators) > 0 {
			sep := separators[len(separators)-1]
			if sep != "" && isSpecialNumEnd(sep[0]) {
				sep = skipFirstToken(sep)
			}
			a = append(a, sep)
			separators = separators[:len(separators)-1]
		}
	case patternMatcherOptionPrefix:
		if len(separators) == 1 {
			sep := skipLastToken(separators[0])
			a = append(a, sep)
			separators = nil
			break
		}

		sep := separators[0]
		if sep != "" && isSpecialNumStart(sep[len(sep)-1]) {
			sep = skipLastToken(sep)
		}
		a = append(a, sep)
		separators = separators[1:]

		if len(separators) > 0 {
			sep := skipLastToken(separators[len(separators)-1])
			if sep != "" && isSpecialNumEnd(sep[0]) {
				sep = skipFirstToken(sep)
			}
			a = append(a, sep)
			separators = separators[:len(separators)-1]
		}
	case patternMatcherOptionSuffix:
		if len(separators) == 1 {
			sep := skipFirstToken(separators[0])
			a = append(a, sep)
			separators = nil
			break
		}

		sep := skipFirstToken(separators[0])
		if sep != "" && isSpecialNumStart(sep[len(sep)-1]) {
			sep = skipLastToken(sep)
		}
		a = append(a, sep)
		separators = separators[1:]

		if len(separators) > 0 {
			sep := separators[len(separators)-1]
			if sep != "" && isSpecialNumEnd(sep[0]) {
				sep = skipFirstToken(sep)
			}
			a = append(a, sep)
			separators = separators[:len(separators)-1]
		}
	default:
		logger.Panicf("BUG: unexpected pmo=%d", fp.pm.pmo)
	}

	for _, sep := range separators {
		if sep != "" && isSpecialNumEnd(sep[0]) {
			sep = skipFirstToken(sep)
		}
		if sep != "" && isSpecialNumStart(sep[len(sep)-1]) {
			sep = skipLastToken(sep)
		}
		a = append(a, sep)
	}

	fp.tokens = tokenizeStrings(nil, a)
	fp.tokensHashes = appendTokensHashes(nil, fp.tokens)
}

func (fp *filterPatternMatch) matchRow(fields []Field) bool {
	v := getFieldValueByName(fields, fp.fieldName)
	return fp.pm.Match(v)
}

func (fp *filterPatternMatch) applyToBlockResult(br *blockResult, bm *bitmap) {
	c := br.getColumnByName(fp.fieldName)
	if c.isConst {
		v := c.valuesEncoded[0]
		if !fp.pm.Match(v) {
			bm.resetBits()
		}
		return
	}
	if c.isTime {
		fp.matchColumnGeneric(br, bm, c)
		return
	}

	switch c.valueType {
	case valueTypeString:
		fp.matchColumnGeneric(br, bm, c)
	case valueTypeDict:
		bb := bbPool.Get()
		for _, v := range c.dictValues {
			c := byte(0)
			if fp.pm.Match(v) {
				c = 1
			}
			bb.B = append(bb.B, c)
		}
		valuesEncoded := c.getValuesEncoded(br)
		bm.forEachSetBit(func(idx int) bool {
			n := valuesEncoded[idx][0]
			return bb.B[n] == 1
		})
		bbPool.Put(bb)
	case valueTypeUint8:
		fp.matchColumnGeneric(br, bm, c)
	case valueTypeUint16:
		fp.matchColumnGeneric(br, bm, c)
	case valueTypeUint32:
		fp.matchColumnGeneric(br, bm, c)
	case valueTypeUint64:
		fp.matchColumnGeneric(br, bm, c)
	case valueTypeInt64:
		fp.matchColumnGeneric(br, bm, c)
	case valueTypeFloat64:
		fp.matchColumnGeneric(br, bm, c)
	case valueTypeIPv4:
		fp.matchColumnGeneric(br, bm, c)
	case valueTypeTimestampISO8601:
		fp.matchColumnGeneric(br, bm, c)
	default:
		logger.Panicf("FATAL: unknown valueType=%d", c.valueType)
	}
}

func (fp *filterPatternMatch) matchColumnGeneric(br *blockResult, bm *bitmap, c *blockResultColumn) {
	values := c.getValues(br)
	bm.forEachSetBit(func(idx int) bool {
		return fp.pm.Match(values[idx])
	})
}

func (fp *filterPatternMatch) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
	fieldName := fp.fieldName

	// Verify whether fp matches const column
	v := bs.getConstColumnValue(fieldName)
	if v != "" {
		if !fp.pm.Match(v) {
			bm.resetBits()
		}
		return
	}

	// Verify whether fp matches other columns
	ch := bs.getColumnHeader(fieldName)
	if ch == nil {
		// Fast path - there are no matching columns.
		if !fp.pm.Match("") {
			bm.resetBits()
		}
		return
	}

	tokens := fp.getTokensHashes()

	switch ch.valueType {
	case valueTypeString:
		if !matchBloomFilterAllTokens(bs, ch, tokens) {
			bm.resetBits()
			return
		}
		visitValues(bs, ch, bm, func(v string) bool {
			return fp.pm.Match(v)
		})
	case valueTypeDict:
		bb := bbPool.Get()
		for _, v := range ch.valuesDict.values {
			c := byte(0)
			if fp.pm.Match(v) {
				c = 1
			}
			bb.B = append(bb.B, c)
		}
		matchEncodedValuesDict(bs, ch, bm, bb.B)
		bbPool.Put(bb)
	case valueTypeUint8:
		if !matchBloomFilterAllTokens(bs, ch, tokens) {
			bm.resetBits()
			return
		}
		bb := bbPool.Get()
		visitValues(bs, ch, bm, func(v string) bool {
			s := toUint8String(bs, bb, v)
			return fp.pm.Match(s)
		})
		bbPool.Put(bb)
	case valueTypeUint16:
		if !matchBloomFilterAllTokens(bs, ch, tokens) {
			bm.resetBits()
			return
		}
		bb := bbPool.Get()
		visitValues(bs, ch, bm, func(v string) bool {
			s := toUint16String(bs, bb, v)
			return fp.pm.Match(s)
		})
		bbPool.Put(bb)
	case valueTypeUint32:
		if !matchBloomFilterAllTokens(bs, ch, tokens) {
			bm.resetBits()
			return
		}
		bb := bbPool.Get()
		visitValues(bs, ch, bm, func(v string) bool {
			s := toUint32String(bs, bb, v)
			return fp.pm.Match(s)
		})
		bbPool.Put(bb)
	case valueTypeUint64:
		if !matchBloomFilterAllTokens(bs, ch, tokens) {
			bm.resetBits()
			return
		}
		bb := bbPool.Get()
		visitValues(bs, ch, bm, func(v string) bool {
			s := toUint64String(bs, bb, v)
			return fp.pm.Match(s)
		})
		bbPool.Put(bb)
	case valueTypeInt64:
		if !matchBloomFilterAllTokens(bs, ch, tokens) {
			bm.resetBits()
			return
		}
		bb := bbPool.Get()
		visitValues(bs, ch, bm, func(v string) bool {
			s := toInt64String(bs, bb, v)
			return fp.pm.Match(s)
		})
		bbPool.Put(bb)
	case valueTypeFloat64:
		if !matchBloomFilterAllTokens(bs, ch, tokens) {
			bm.resetBits()
			return
		}
		bb := bbPool.Get()
		visitValues(bs, ch, bm, func(v string) bool {
			s := toFloat64String(bs, bb, v)
			return fp.pm.Match(s)
		})
		bbPool.Put(bb)
	case valueTypeIPv4:
		if !matchBloomFilterAllTokens(bs, ch, tokens) {
			bm.resetBits()
			return
		}
		bb := bbPool.Get()
		visitValues(bs, ch, bm, func(v string) bool {
			s := toIPv4String(bs, bb, v)
			return fp.pm.Match(s)
		})
		bbPool.Put(bb)
	case valueTypeTimestampISO8601:
		if !matchBloomFilterAllTokens(bs, ch, tokens) {
			bm.resetBits()
			return
		}
		bb := bbPool.Get()
		visitValues(bs, ch, bm, func(v string) bool {
			s := toTimestampISO8601String(bs, bb, v)
			return fp.pm.Match(s)
		})
		bbPool.Put(bb)
	default:
		logger.Panicf("FATAL: %s: unknown valueType=%d", bs.partPath(), ch.valueType)
	}
}
