common/repository/statistics/click_statistics.go

package statistics

import (
	"context"
	"database/sql"
	"fmt"
	"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, keyos,
    ctxuserip, ctxuserport, keydomain, keypath, ctxquiz, ctxreferrer
FROM (SELECT
            event_time, ctxsession, ctxquizid, ctxquestionid, ctxidint, message, keyos,
            ctxuserip, ctxuserport, keydomain, keypath, ctxquiz, ctxreferrer,
            row_number() OVER (PARTITION BY ctxsession ORDER BY event_time DESC) as row_num
        FROM statistics
        WHERE message IN ('InfoQuizOpen', 'InfoAnswer', 'InfoResult') AND event_level = 'info') AS sorted
WHERE row_num = 1;
`
	_, err := s.conn.ExecContext(ctx, query)
	if err != nil {
		return err
	}
	return nil
}

type Statistic struct {
	Count      int64
	QuestionID int64
}

type PipeLineStatsResp [][]Statistic

// todo нужно исключить множества из подмножеств, пока что получаются все уникальные векторы респондентов по опросу
// пример:
//"[0, 116783, 116810]"
//"[0, 116783, 116798]"
//"[0, 116783, 116798, 116831]"
//"[0, 116783, 116810, 116849]"
//[0]
//"[0, 116783]"
//"[0, 116783, 116810, 116843]"

//SELECT DISTINCT final FROM (
//SELECT groupArray(ctxquestionid) AS final FROM (SELECT DISTINCT f.ctxsession, a.ctxquestionid
//FROM (SELECT ctxsession, max(event_time) AS max_time
//FROM statistics WHERE ctxquizid = 26276 GROUP BY ctxsession ) f
//JOIN ( SELECT ctxsession, ctxquestionid, event_time
//FROM statistics WHERE ctxquizid = 26276 ) a ON f.ctxsession = a.ctxsession)
//GROUP BY ctxsession);

func (s *StatisticClick) getFunnel(ctx context.Context, quizID int64, from uint64, to uint64) (map[int64][]string, error) {
	// берем из матвью все что принадлежит quizID в указанном интервале времени
	// выбираем самыые поздние по роу набер - 1
	// группируем по ид вопроса и для каждого ид вопроса формируем массив сессий которые были последнимим для этого вопроса
	// выбираем только те где длина массива = 1
	query := `
	SELECT ctxquestionid, arrayJoin(endsession) AS session
	FROM (SELECT  ctxquestionid,   groupArray(ctxsession) AS endsession
	    FROM (SELECT ctxsession,ctxquestionid,row_number() OVER (PARTITION BY ctxsession ORDER BY event_time DESC) AS row_num
	        FROM mv_last_answers_events WHERE ctxquizid = ? AND event_time BETWEEN ? AND ?
	    ) AS rows
	    WHERE row_num = 1 GROUP BY ctxquestionid
	) AS group_sessions WHERE length(endsession) = 1;
	`

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

	funnel := make(map[int64][]string)

	for rows.Next() {
		var questionID int64
		var sessionID string
		err := rows.Scan(&questionID, &sessionID)
		if err != nil {
			return nil, err
		}
		funnel[questionID] = append(funnel[questionID], sessionID)
	}

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

	return funnel, nil
}

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

	// получили id вопросов воронок где массив состоит из 1 элемента
	funnel, err := s.getFunnel(ctx, quizID, from, to)
	if err != nil {
		return nil, err
	}
	var idS []int64
	for queID := range funnel {
		idS = append(idS, queID)
	}

	if len(idS) == 0 {
		return nil, nil
	}
	// тут считаем количество ответов на эти вопросы по уникальным сессиям
	sesCount, err := s.countSession(ctx, quizID, from, to, idS)
	if err != nil {
		return nil, err
	}

	for questionID := range funnel {
		if sessionCount, ok := sesCount[questionID]; ok {
			pipeline := []Statistic{
				{
					QuestionID: questionID,
					Count:      sessionCount,
				},
			}
			pipelines = append(pipelines, 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
}