common/repository/statistics/click_statistics.go

package statistics

import (
	"context"
	"database/sql"
	"fmt"
	"penahub.gitlab.yandexcloud.net/backend/quiz/common.git/utils"
	"sort"
	"strings"
)

type DepsClick struct {
	Conn *sql.DB
}

type StatisticClick struct {
	conn *sql.DB
}

func NewClickStatistic(ctx context.Context, deps DepsClick) (*StatisticClick, error) {
	s := &StatisticClick{
		conn: deps.Conn,
	}

	err := s.checkMW(ctx)
	if err != nil {
		fmt.Println("error check material view existing", err)
		return nil, err
	}

	return s, nil
}

func (s *StatisticClick) checkMW(ctx context.Context) error {
	query := `
CREATE MATERIALIZED VIEW IF NOT EXISTS mv_last_answers_events
ENGINE = MergeTree()
ORDER BY (ctxsession, event_time) POPULATE AS
SELECT
    event_time, ctxsession, ctxquizid, ctxquestionid, ctxidint, message,ctxquiz
FROM statistics
WHERE message IN ('InfoQuizOpen', 'InfoAnswer', 'InfoResult') AND event_level = 'info';
`
	_, err := s.conn.ExecContext(ctx, query)
	if err != nil {
		return err
	}
	return nil
}

type Statistic struct {
	Count      int64
	QuestionID int64
}

type PipeLineStatsResp map[int64][]Statistic

// пример:
//"[0, 116783, 116810]"
//"[0, 116783, 116798]"
//"[0, 116783, 116798, 116831]"
//"[0, 116783, 116810, 116849]"
//[0]
//"[0, 116783]"
//"[0, 116783, 116810, 116843]"

//SELECT DISTINCT last_que, reversed
//FROM ( SELECT groupArray(ctxquestionid) AS reversed, arraySlice(arrayReverse(groupArray(ctxquestionid)), 1, 1)[1] AS last_que
//FROM statistics WHERE ctxquizid = 26276  GROUP BY ctxsession ) AS sub;

func (s *StatisticClick) getFunnel(ctx context.Context, quizID int64, from uint64, to uint64) (map[int64][]int64, error) {
	query := `
	SELECT DISTINCT last_que, reversed
	FROM ( SELECT groupUniqArray(ctxquestionid) AS reversed, arraySlice(arrayReverse(groupArray(ctxquestionid)), 1, 1)[1] AS last_que
         FROM mv_last_answers_events WHERE ctxquizid = ? AND event_time BETWEEN ? AND ?  GROUP BY ctxsession ) AS sub;
	`

	rows, err := s.conn.QueryContext(ctx, query, quizID, from, to)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	funnel := make(map[int64][]int64)
	for rows.Next() {
		var lastQue int64
		var reversed []int64
		if err := rows.Scan(&lastQue, &reversed); err != nil {
			return nil, err
		}
		funnel[lastQue] = reversed
	}

	if err := rows.Err(); err != nil {
		return nil, err
	}

	result := make(map[int64][]int64)
	keys := make([]int64, 0, len(funnel))
	for key := range funnel {
		keys = append(keys, key)
	}
	sort.Slice(keys, func(i, j int) bool {
		return keys[i] < keys[j]
	})

	for _, lastQue := range keys {
		reversed := funnel[lastQue]
		found := false

		for _, otherLastQue := range keys {
			if otherLastQue != lastQue {
				otherReversed := funnel[otherLastQue]
				sort.Slice(otherReversed, func(i, j int) bool {
					return otherReversed[i] < otherReversed[j]
				})
				index := utils.BinarySearch(lastQue, otherReversed)
				if index {
					found = true
					break
				}
			}
		}

		if !found {
			result[lastQue] = reversed
		}
	}

	return result, nil
}

func (s *StatisticClick) GetPipelinesStatistics(ctx context.Context, quizID int64, from uint64, to uint64) (PipeLineStatsResp, error) {
	pipelines := make(PipeLineStatsResp)

	funnel, err := s.getFunnel(ctx, quizID, from, to)
	if err != nil {
		return nil, err
	}
	for lastQue, idS := range funnel {
		sesCount, err := s.countSession(ctx, quizID, from, to, idS)
		if err != nil {
			return nil, err
		}
		for _, queID := range idS {
			if sessionCount, ok := sesCount[queID]; ok {
				pipeline := Statistic{
					QuestionID: queID,
					Count:      sessionCount,
				}
				pipelines[lastQue] = append(pipelines[lastQue], pipeline)
			}
		}
	}
	return pipelines, nil
}

func (s *StatisticClick) countSession(ctx context.Context, quizID int64, from uint64, to uint64, questionIDs []int64) (map[int64]int64, error) {
	placeholders := make([]string, len(questionIDs))
	args := make([]interface{}, len(questionIDs)+3)
	args[0] = quizID
	for i, id := range questionIDs {
		placeholders[i] = "?"
		args[i+1] = id
	}
	args[len(args)-2] = from
	args[len(args)-1] = to

	query := fmt.Sprintf(`
	SELECT ctxquestionid, COUNT(DISTINCT ctxsession) AS session_count
	FROM statistics
	WHERE ctxquizid = ? AND ctxquestionid IN (%s) AND event_time BETWEEN ? AND ?
	GROUP BY ctxquestionid;
	`, strings.Join(placeholders, ","))

	rows, err := s.conn.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	counts := make(map[int64]int64)

	for rows.Next() {
		var questionID int64
		var count int64
		err := rows.Scan(&questionID, &count)
		if err != nil {
			return nil, err
		}
		counts[questionID] = count
	}

	if err := rows.Err(); err != nil {
		return nil, err
	}

	return counts, nil
}