core/webapi/Search Job.go

/*
File Name:  Search Job.go
Copyright:  2021 Peernet Foundation s.r.o.
Author:     Peter Kleissner
*/

package webapi

import (
	"sort"
	"sync"
	"time"

	"github.com/PeernetOfficial/core/blockchain"
	"github.com/google/uuid"
)

// SearchFilter allows to filter search results based on the criteria.
type SearchFilter struct {
	IsDates    bool      // Whether the from/to dates are valid, both are required.
	DateFrom   time.Time // Optional date from
	DateTo     time.Time // Optional date to
	FileType   int       // File type such as binary, text document etc. See core.TypeX. -1 = not used.
	FileFormat int       // File format such as PDF, Word, Ebook, etc. See core.FormatX. -1 = not used.
	Sort       int       // Sort order. See SortX.
	SizeMin    int       // Min file size in bytes. -1 = not used.
	SizeMax    int       // Max file size in bytes. -1 = not used.
}

// SearchJob is a collection of search jobs
type SearchJob struct {
	// input settings
	id        uuid.UUID     // The job id
	timeout   time.Duration // timeout set for all searches
	maxResult int           // max results user-facing.

	filtersStart   SearchFilter // Filters when starting the search. They cannot be changed later on. Any incoming file is checked against them, even if there are different runtime filters.
	filtersRuntime SearchFilter // Runtime Filters. They allow filtering results after they were received.

	// File statistics (filters are ignored) of returned results. Map value is always count of files.
	stats struct {
		sync.RWMutex                   // Synced access to maps
		date         map[time.Time]int // Files per day (rounded down to midnight)
		fileType     map[uint8]int     // Files per File Type
		fileFormat   map[uint16]int    // Files per File Format
		total        int               // Total count of files
	}

	// -- result data --

	// runtime data
	//clients        []*SearchClient // all search clients
	clientsMutex sync.Mutex // mutex for manipulating client list

	// List of files found but not yet returned via API to the caller. They are subject to sorting.
	Files       []*apiFile
	requireSort bool // if Files requires sort before returning the results

	// FreezeFiles is a list of files that were already finally delivered via the API. They may NOT change in sorting.
	FreezeFiles []*apiFile

	// List of all files. Does not change based on sorting or runtime filters. This list only gets expanded.
	AllFiles []*apiFile

	ResultSync sync.Mutex // ResultSync ensures unique access to the file results

	currentOffset int // for always getting the next results
}

// CreateSearchJob creates a new search job and adds it to the lookup list.
// Timeout and MaxResults must be set and must not be 0.
func CreateSearchJob(Timeout time.Duration, MaxResults int, Filter SearchFilter) (job *SearchJob) {
	job = &SearchJob{}
	job.id = uuid.New()
	job.timeout = Timeout
	job.maxResult = MaxResults
	job.filtersStart = Filter
	job.filtersRuntime = Filter // initialize the runtime filters as the same

	job.stats.date = make(map[time.Time]int)
	job.stats.fileType = make(map[uint8]int)
	job.stats.fileFormat = make(map[uint16]int)

	// add to the list of jobs
	allJobsMutex.Lock()
	allJobs[job.id] = job
	allJobsMutex.Unlock()

	return
}

// ReturnResult returns the selected results.
func (job *SearchJob) ReturnResult(Offset, Limit int) (Result []*apiFile) {
	if Limit == 0 {
		return Result
	}

	job.ResultSync.Lock()
	defer job.ResultSync.Unlock()

	// serve files from frozen list?
	if Offset < len(job.FreezeFiles) {
		countCopy := len(job.FreezeFiles) - Offset
		if countCopy > Limit {
			countCopy = Limit
		}
		Result = job.FreezeFiles[Offset : Offset+countCopy]
		Limit -= countCopy
		Offset = 0
	} else {
		Offset -= len(job.FreezeFiles)
	}

	if Limit == 0 {
		return Result
	}

	// go through the live results and fill the list
	if Offset >= len(job.Files) { // offset wants to skip entire queue?
		job.FreezeFiles = append(job.FreezeFiles, job.Files...)
		job.Files = nil
		return Result
	}

	// check if a sort is required before using this queue
	if job.requireSort {
		job.requireSort = false

		job.Files = SortFiles(job.Files, job.filtersRuntime.Sort)
	}

	// set the amount of files to copy
	countCopy := len(job.Files) - Offset
	if countCopy > Limit {
		countCopy = Limit
	}

	// copy the results and freeze them
	Result = append(Result, job.Files[Offset:Offset+countCopy]...)

	// note that freeze disregards the offset, it has to freeze any elements before!
	job.FreezeFiles = append(job.FreezeFiles, job.Files[:Offset+countCopy]...)
	job.Files = job.Files[Offset+countCopy:]

	//Limit -= countCopy

	return Result
}

// ReturnNext returns the next results. Call must be serialized.
func (job *SearchJob) ReturnNext(Limit int) (Result []*apiFile) {
	Result = job.ReturnResult(job.currentOffset, Limit)
	job.currentOffset += len(Result)

	return
}

// PeekResult returns the selected results but will not change any frozen files or impact auto offset
func (job *SearchJob) PeekResult(Offset, Limit int) (Result []*apiFile) {
	job.ResultSync.Lock()
	defer job.ResultSync.Unlock()

	// serve files from frozen list?
	if Offset < len(job.FreezeFiles) {
		countCopy := len(job.FreezeFiles) - Offset
		if countCopy > Limit {
			countCopy = Limit
		}
		Result = job.FreezeFiles[Offset : Offset+countCopy]
		Limit -= countCopy
		Offset = 0
	} else {
		Offset -= len(job.FreezeFiles)
	}

	if Limit == 0 || Offset >= len(job.Files) { // offset wants to skip entire queue?
		return Result
	}

	// check if a sort is required before using this queue
	if job.requireSort {
		job.requireSort = false

		job.Files = SortFiles(job.Files, job.filtersRuntime.Sort)
	}

	countCopy := len(job.Files) - Offset
	if countCopy > Limit {
		countCopy = Limit
	}

	// copy the results
	Result = append(Result, job.Files[Offset:Offset+countCopy]...)

	return Result
}

// RuntimeFilter allows to apply filters at runtime to search jobs that already started. To remove the filters, call this function without the filters set.
func (job *SearchJob) RuntimeFilter(Filter SearchFilter) {
	job.ResultSync.Lock()
	defer job.ResultSync.Unlock()

	job.filtersRuntime = Filter

	// FreezeFiles and current offset is reset
	job.FreezeFiles = nil
	job.currentOffset = 0

	// files remain in AllFiles, but Files needs to be filtered based on the new filter
	job.Files = []*apiFile{}
	job.requireSort = false // Sorting is done immediately below

	// set Files based on AllFiles with the filter
	for m := range job.AllFiles {
		// only append if filter is matching
		if job.isFileFiltered(job.AllFiles[m]) {
			job.Files = append(job.Files, job.AllFiles[m])
		}

		// sort, if a sort order is defined
		if job.filtersRuntime.Sort > 0 {
			job.Files = SortFiles(job.Files, job.filtersRuntime.Sort)
		}
	}
}

// isFileFiltered returns true if the file conforms to the runtime filter. If there is no runtime filter, it always returns true.
func (job *SearchJob) isFileFiltered(file *apiFile) bool {
	if job.filtersRuntime.FileType >= 0 && file.Type != uint8(job.filtersRuntime.FileType) {
		return false
	}

	if job.filtersRuntime.FileFormat >= 0 && file.Format != uint16(job.filtersRuntime.FileFormat) {
		return false
	}

	// Note: If the date is not available in the file, it will be filtered out. Since this is the mapped Shared Date this should normally not occur though.
	if job.filtersRuntime.IsDates && (file.Date.IsZero() || file.Date.Before(job.filtersRuntime.DateFrom) || file.Date.After(job.filtersRuntime.DateTo)) {
		return false
	}

	if job.filtersRuntime.SizeMin >= 0 && file.Size < uint64(job.filtersRuntime.SizeMin) || job.filtersRuntime.SizeMax >= 0 && file.Size > uint64(job.filtersRuntime.SizeMax) {
		return false
	}

	return true
}

// SortFiles sorts a list of files. It returns a sorted list. 0 = no sorting, 1 = Relevance ASC, 2 = Relevance DESC, 3 = Date ASC, 4 = Date DESC, 5 = Name ASC, 6 = Name DESC
func SortFiles(files []*apiFile, Sort int) (sorted []*apiFile) {
	switch Sort {
	case SortRelevanceAsc:
		sort.SliceStable(files, func(i, j int) bool { return files[i].Date.Before(files[j].Date) }) // first as date for secondary sorting
		//sort.SliceStable(files, func(i, j int) bool { return files[i].Score < files[j].Score }) // TODO
	case SortRelevanceDec:
		sort.SliceStable(files, func(i, j int) bool { return files[j].Date.Before(files[i].Date) }) // first as date for secondary sorting
		//sort.SliceStable(files, func(i, j int) bool { return files[i].Score > files[j].Score }) // TODO

	case SortDateAsc:
		sort.SliceStable(files, func(i, j int) bool { return files[i].Date.Before(files[j].Date) })
	case SortDateDesc:
		sort.SliceStable(files, func(i, j int) bool { return files[j].Date.Before(files[i].Date) })

	case SortNameAsc:
		sort.SliceStable(files, func(i, j int) bool { return files[i].Name < files[j].Name })
	case SortNameDesc:
		sort.SliceStable(files, func(i, j int) bool { return files[i].Name > files[j].Name })

	case SortSizeAsc:
		sort.SliceStable(files, func(i, j int) bool { return files[i].Size < files[j].Size })
	case SortSizeDesc:
		sort.SliceStable(files, func(i, j int) bool { return files[i].Size > files[j].Size })

	case SortSharedByCountAsc:
		sort.SliceStable(files, func(i, j int) bool {
			return files[i].GetMetadata(blockchain.TagSharedByCount).Number < files[j].GetMetadata(blockchain.TagSharedByCount).Number
		})
	case SortSharedByCountDesc:
		sort.SliceStable(files, func(i, j int) bool {
			return files[i].GetMetadata(blockchain.TagSharedByCount).Number > files[j].GetMetadata(blockchain.TagSharedByCount).Number
		})

	}

	return files
}

// IsSearchResults checks if search results may be expected (either files are in queue or a search is running)
func (job *SearchJob) IsSearchResults() bool {
	// check for any available results. Do not use any lock here as this is read only.
	return len(job.Files) > 0 || !job.IsTerminated()
}

// isFileReceived checks if a file was already received, preventing double results
func (job *SearchJob) isFileReceived(id uuid.UUID) (exists bool) {
	// Future: A map would be likely faster than iterating over all results.
	for m := range job.AllFiles {
		if id == job.AllFiles[m].ID {
			return true
		}
	}

	return false
}

// ---- job list management ----
var (
	allJobs      map[uuid.UUID]*SearchJob = make(map[uuid.UUID]*SearchJob)
	allJobsMutex sync.RWMutex
)

// Remove removes the structure from the list. Terminate should be called before. Unless the search is manually removed, it stays forever in the list.
func (job *SearchJob) Remove() {
	allJobsMutex.Lock()
	delete(allJobs, job.id) // delete is safe to call multiple times, so auto-removal and manual one are fine and need no syncing
	allJobsMutex.Unlock()
}

// RemoveDefer removes the search job after a given time after all searches are terminated. This can be used for automated time delayed removal. Do not create additional search clients after deferal removing.
func (job *SearchJob) RemoveDefer(Duration time.Duration) {
	go func() {
		// for _, client := range job.clients {
		// 	<-client.TerminateSignal
		// }

		<-time.After(Duration)
		job.Remove()
	}()
}

// JobLookup looks up a job. Returns nil if not found.
func JobLookup(id uuid.UUID) (job *SearchJob) {
	allJobsMutex.RLock()
	job = allJobs[id]
	allJobsMutex.RUnlock()

	return job
}

// IsTerminated checks if all searches are finished. The job itself does not record a termination signal.
func (job *SearchJob) IsTerminated() bool {
	job.clientsMutex.Lock()
	defer job.clientsMutex.Unlock()

	// for n := range job.clients {
	// 	if !job.clients[n].IsTerminated {
	// 		return false
	// 	}
	// }

	return false
}

// Terminate terminates all searches
func (job *SearchJob) Terminate() {
	job.clientsMutex.Lock()
	defer job.clientsMutex.Unlock()

	// for n := range job.clients {
	// 	if !job.clients[n].IsTerminated {
	// 		job.clients[n].Terminate(true)
	// 	}
	// }
}

// WaitTerminate waits until all search clients are terminated. Do not create additional search clients after calling this function.
func (job *SearchJob) WaitTerminate() {
	//for _, client := range job.clients {
	//	<-client.TerminateSignal
	//}
}

// ---- statistics ----

// SearchStatisticRecordDay is a single record containing date info.
type SearchStatisticRecordDay struct {
	Date  time.Time `json:"date"`  // The day (which covers the full 24 hours). Always rounded down to midnight.
	Count int       `json:"count"` // Count of files.
}

// SearchStatisticRecord is a single record.
type SearchStatisticRecord struct {
	Key   int `json:"key"`   // Key index. The exact meaning depends on where this structure is used.
	Count int `json:"count"` // Count of files for the given key
}

// SearchStatisticData contains statistics on search results.
type SearchStatisticData struct {
	Date       []SearchStatisticRecordDay `json:"date"`       // Files per date
	FileType   []SearchStatisticRecord    `json:"filetype"`   // Files per file type
	FileFormat []SearchStatisticRecord    `json:"fileformat"` // Files per file format
	Total      int                        `json:"total"`      // Total count of files
}

// Statistics generates statistics on all results, regardless of runtime filters.
func (job *SearchJob) Statistics() (result SearchStatisticData) {
	job.stats.RLock()
	defer job.stats.RUnlock()

	result.Total = job.stats.total

	// Files per date. Sort dates to date ASC.
	for key, value := range job.stats.date {
		result.Date = append(result.Date, SearchStatisticRecordDay{Date: key, Count: value})
	}
	sort.SliceStable(result.Date, func(i, j int) bool { return result.Date[i].Date.Before(result.Date[j].Date) })

	// File Type and Format
	for key, value := range job.stats.fileType {
		result.FileType = append(result.FileType, SearchStatisticRecord{Key: int(key), Count: value})
	}
	for key, value := range job.stats.fileFormat {
		result.FileFormat = append(result.FileFormat, SearchStatisticRecord{Key: int(key), Count: value})
	}

	return
}

// statsAdd counts the files in the statistics
func (job *SearchJob) statsAdd(files ...*apiFile) {
	job.stats.Lock()
	defer job.stats.Unlock()

	for _, file := range files {
		// Use file's Date field if available.
		if !file.Date.IsZero() {
			// Files per day
			date := file.Date.Truncate(24 * time.Hour)
			countDate := job.stats.date[date]
			countDate++
			job.stats.date[date] = countDate
		}

		// File Type and Format
		countType := job.stats.fileType[file.Type]
		countType++
		job.stats.fileType[file.Type] = countType

		countFormat := job.stats.fileFormat[file.Format]
		countFormat++
		job.stats.fileFormat[file.Format] = countFormat
	}

	job.stats.total += len(files)
}

// ---- actual search & retrieving results ----

// appendSearchClient appends a search client to a job

func (job *SearchJob) SearchAway() {
	job.clientsMutex.Lock()
	defer job.clientsMutex.Unlock()

	// TODO
}

//func (job *SearchJob) appendSearchClient(search *dht.SearchClient) {
//}

//func (job *SearchJob) receiveClientResults(client *dht.SearchClient) {
//}