Вручную сосквошил chMigrate, чтобы понимать происходящее. Ничего не понял толком.

Тут сделаны миграции и хендлеры для того чтобы выдавать ltv для промокода и недо партнёрка по переходам через ссылку в лого  опроса

dal/ch_schema/000001_init.down.sql

@@ -1,5 +1,8 @@
+DROP MATERIALIZED VIEW IF EXISTS promo_view;
+DROP MATERIALIZED VIEW IF EXISTS user_regs_view;
+DROP MATERIALIZED VIEW IF EXISTS ltv_view;
 DROP VIEW IF EXISTS view_respondent_paths;
 DROP VIEW IF EXISTS view_pipelines_signs;
 DROP VIEW IF EXISTS view_last_answers;
-DROP MATERIALIZED VIEW IF EXISTS mv_answers;
+DROP MATERIALIZED VIEW IF EXISTS mv_quiz_open;
-DROP MATERIALIZED VIEW if exists mv_last_answers_events;
+DROP MATERIALIZED VIEW IF EXISTS mv_answers;

dal/ch_schema/000001_init.up.sql

@@ -1,3 +1,4 @@
+DROP MATERIALIZED VIEW if exists mv_last_answers_events;
 CREATE MATERIALIZED VIEW IF NOT EXISTS mv_answers
             ENGINE = MergeTree() ORDER BY event_time
             POPULATE AS
@@ -10,6 +11,14 @@ WHERE (message IN ('InfoQuizOpen', 'InfoAnswer', 'InfoResult'))
   AND (event_level = 'info')
 LIMIT 1 BY ctxquestionid, ctxsession;
 
+CREATE MATERIALIZED VIEW IF NOT EXISTS mv_quiz_open
+    ENGINE = MergeTree()
+    ORDER BY event_time POPULATE AS
+SELECT ctxquizid AS quizid,ctxquestionid AS questionid, ctxsession AS session, event_time
+FROM default.statistics
+WHERE message = 'InfoContactForm'AND event_level = 'info'
+LIMIT 1 BY ctxquestionid, ctxsession;
+
 CREATE VIEW IF NOT EXISTS view_last_answers AS
 SELECT last_quesion,
     any(last_quiz)                     AS quiz,
@@ -40,3 +49,43 @@ SELECT session,
 FROM default.mv_answers
 GROUP BY session;
 
+-- тут хранится информация по оплатах пользователей именно успешных, успешность показывает keysuccess = 1
+CREATE MATERIALIZED VIEW IF NOT EXISTS ltv_view
+            ENGINE = MergeTree()
+                ORDER BY (ctxuserid)
+            POPULATE AS
+SELECT create_time, -- время создания
+       ctxuserid, -- id пользователя
+       ctxaccountid, -- id oid в монго
+       ctxprice, -- цена корзины после подсчета скидки
+       ctxrowprice -- цена корзины до скидки
+FROM default.statistics
+WHERE keysuccess = 1 and message = 'InfoPayCart'AND event_level = 'info';
+
+
+CREATE MATERIALIZED VIEW IF NOT EXISTS user_regs_view
+            ENGINE = MergeTree()
+                ORDER BY (ctxuserid)
+            POPULATE AS
+SELECT create_time, -- время создания
+       ctxuserid, -- id пользователя
+       ctxaccountid, -- oid акканута в монго
+       keyfromsource, -- название сервиса откуда пришел, интересно пока что только quiz
+       keyfromid, -- quiz id откуда с которого пришел пользователь
+       keyfrompartner, -- id пользователя который является партнером (тот кто привел)
+       ctxemail, -- почта пользоователя
+       ctxlogin -- логин пользователя
+FROM default.statistics
+WHERE message = 'InfoCreateAccount'AND event_level = 'info';
+
+CREATE MATERIALIZED VIEW IF NOT EXISTS promo_view
+            ENGINE = MergeTree()
+                ORDER BY (ctxuserid)
+            POPULATE AS
+SELECT create_time, --время создания
+       ctxid, -- id промо в монге
+       ctxuserid, -- id пользователя который активировал
+       ctxcode, -- кодовое слово (заполняется если по нему была активация)
+       ctxpromocodeid -- фастлинк (заполняется если по нему была активация)
+FROM default.statistics
+WHERE message IN ( 'InfoPromocodeActivated', 'InfoFastlinkActivated') AND event_level = 'info';

repository/statistics/click_statistics.go

@@ -3,6 +3,8 @@ package statistics
 import (
 	"context"
 	"database/sql"
+	"fmt"
+	"time"
 )
 
 type DepsClick struct {
@@ -19,6 +21,11 @@ func NewClickStatistic(deps DepsClick) *StatisticClick {
 	}
 }
 
+type PipelineAndFormStatistic struct {
+	PipelineStatistic    PipeLineStatsResp
+	ContactFormStatistic map[int64]int
+}
+
 type Statistic struct {
 	Count      int64
 	QuestionID int64
@@ -26,7 +33,7 @@ type Statistic struct {
 
 type PipeLineStatsResp map[int64][]Statistic
 
-func (s *StatisticClick) GetPipelinesStatistics(ctx context.Context, quizID int64, from uint64, to uint64) (PipeLineStatsResp, error) {
+func (s *StatisticClick) GetPipelinesStatistics(ctx context.Context, quizID int64, from uint64, to uint64) (*PipelineAndFormStatistic, error) {
 	pipelines := make(PipeLineStatsResp)
 
 	query := `
@@ -64,5 +71,204 @@ func (s *StatisticClick) GetPipelinesStatistics(ctx context.Context, quizID int6
 		return nil, err
 	}
 
-	return pipelines, nil
+	contactQuery := `
+        SELECT questionid, count() FROM mv_quiz_open WHERE quizid = ? AND event_time BETWEEN ? AND ?
+        GROUP BY questionid;
+    `
+
+	contactRows, err := s.conn.QueryContext(ctx, contactQuery, quizID, from, to)
+	if err != nil {
+		return nil, err
+	}
+	defer contactRows.Close()
+
+	contactForms := make(map[int64]int)
+	for contactRows.Next() {
+		var questionID int64
+		var count int
+		if err := contactRows.Scan(&questionID, &count); err != nil {
+			return nil, err
+		}
+		contactForms[questionID] = count
+	}
+
+	if err := contactRows.Err(); err != nil {
+		return nil, err
+	}
+
+	return &PipelineAndFormStatistic{
+		PipelineStatistic:    pipelines,
+		ContactFormStatistic: contactForms,
+	}, nil
+}
+
+func (s *StatisticClick) CalculateCustomerLTV(ctx context.Context, req ChStatDeps) (int64, error) {
+	timeFilter := ""
+	if req.From != 0 {
+		fromTime := time.Unix(req.From, 0).UTC().Format("2006-01-02 15:04:05")
+		timeFilter += fmt.Sprintf(" AND create_time >= '%s'", fromTime)
+	}
+	if req.To != 0 {
+		toTime := time.Unix(req.To, 0).UTC().Format("2006-01-02 15:04:05")
+		timeFilter += fmt.Sprintf(" AND create_time <= '%s'", toTime)
+	}
+
+	query := fmt.Sprintf(`
+SELECT avg(lifeTimeInDays) AS averageLTV
+FROM (
+         SELECT
+             (dateDiff('day', min(create_time), max(create_time))) AS lifeTimeInDays
+         FROM ltv_view WHERE 1 = 1 %s  GROUP BY ctxuserid)
+    `, timeFilter)
+
+	var result struct{ AverageLTV float64 }
+	if err := s.conn.QueryRowContext(ctx, query).Scan(&result.AverageLTV); err != nil {
+		return 0, err
+	}
+
+	return int64(result.AverageLTV), nil
+}
+
+type ChStatDeps struct {
+	From int64
+	To   int64
+}
+
+type QuizLogoStats struct {
+	ID     string
+	Regs   int
+	Money  int64
+	Quizes []Quiz
+}
+
+type Quiz struct {
+	QuizID string
+	Regs   uint64
+	Money  int64
+}
+
+func (s *StatisticClick) QuizLogoStat(ctx context.Context, req ChStatDeps) ([]QuizLogoStats, error) {
+	timeFilter := ""
+	if req.From != 0 && req.To != 0 {
+		timeFilter = fmt.Sprintf("AND lv.create_time >= toDate(%d) AND lv.create_time <= toDate(%d)", req.From, req.To)
+	}
+	query := fmt.Sprintf(`
+	SELECT
+    sub.partner AS partner,
+    count() AS regs,
+    sum(sub.rowPrice) AS money,
+    groupArray((sub.fromID, sub.regCount, sub.sumPrice)) AS quizes
+FROM (
+         SELECT
+             uv.keyfrompartner AS partner,
+             uv.keyfromid AS fromID,
+             count() AS regCount,
+             sum(lv.ctxprice) AS sumPrice,
+             sum(lv.ctxprice) AS rowPrice
+         FROM ltv_view lv
+                  JOIN user_regs_view uv ON lv.ctxuserid = uv.ctxuserid
+         WHERE uv.keyfrompartner != '' AND uv.keyfromid != ''
+        	%s
+         GROUP BY uv.keyfrompartner, uv.keyfromid
+         ) AS sub
+GROUP BY sub.partner;
+	`, timeFilter)
+
+	var results []QuizLogoStats
+	rows, err := s.conn.QueryContext(ctx, query)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var stats QuizLogoStats
+		var quizesData [][]interface{}
+
+		err := rows.Scan(&stats.ID, &stats.Regs, &stats.Money, &quizesData)
+		if err != nil {
+			return nil, err
+		}
+
+		var quizes []Quiz
+		for _, q := range quizesData {
+			quiz := Quiz{
+				QuizID: q[0].(string),
+				Regs:   q[1].(uint64),
+				Money:  q[2].(int64),
+			}
+			quizes = append(quizes, quiz)
+		}
+
+		stats.Quizes = quizes
+		results = append(results, stats)
+	}
+
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+
+	return results, nil
+}
+
+type PromoLtvResp struct {
+	Regs  int
+	Money int
+}
+
+func (s *StatisticClick) PromocodeLTV(ctx context.Context, req ChStatDeps) (map[string]PromoLtvResp, error) {
+	timeFilterFirstPromo := ""
+	timeFilterFirstData := ""
+	timeFilterMainQuery := ""
+	if req.From != 0 && req.To != 0 {
+		timeFilterFirstPromo = fmt.Sprintf("AND create_time BETWEEN toDateTime(%d) AND toDateTime(%d)", req.From, req.To)
+		timeFilterFirstData = fmt.Sprintf("AND promo.create_time BETWEEN toDateTime(%d) AND toDateTime(%d)", req.From, req.To)
+		timeFilterMainQuery = fmt.Sprintf("AND ltv.create_time BETWEEN toDateTime(%d) AND toDateTime(%d)", req.From, req.To)
+	}
+
+	query := fmt.Sprintf(`
+	WITH first_promo AS (
+    SELECT
+        ctxuserid, min(create_time) AS first_promo_time
+    FROM promo_view
+    WHERE 1=1 %s
+    GROUP BY ctxuserid
+),
+  first_data AS (
+     SELECT
+         promo.ctxid, promo.ctxuserid, promo.create_time AS first_time
+     FROM
+         promo_view AS promo JOIN first_promo AS first
+         ON promo.ctxuserid = first.ctxuserid AND promo.create_time = first.first_promo_time
+     WHERE 1=1 %s
+  )
+SELECT
+    promo_data.ctxid AS promoID,
+    count(DISTINCT promo_data.ctxuserid) AS regs,
+    sum(if(ltv.create_time > promo_data.first_time %s, ltv.ctxprice, 0)) AS totalSpent
+FROM
+    first_data AS promo_data LEFT JOIN ltv_view AS ltv
+    ON promo_data.ctxuserid = ltv.ctxuserid
+GROUP BY promo_data.ctxid;
+	`, timeFilterFirstPromo, timeFilterFirstData, timeFilterMainQuery)
+
+	rows, err := s.conn.QueryContext(ctx, query)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	results := make(map[string]PromoLtvResp)
+	for rows.Next() {
+		var promoID string
+		var resp PromoLtvResp
+		if err := rows.Scan(&promoID, &resp.Regs, &resp.Money); err != nil {
+			return nil, err
+		}
+		results[promoID] = resp
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return results, nil
 }

repository/statistics/statistics.go

@@ -1,7 +1,6 @@
 package statistics
 
 import (
-	"fmt"
 	"context"
 	"database/sql"
 	"gitea.pena/SQuiz/common/dal/sqlcgen"
@@ -64,10 +63,10 @@ func (r *StatisticsRepository) GetDeviceStatistics(ctx context.Context, req Devi
 }
 
 type GeneralStatsResp struct {
-	Open       map[int64]int64  // количество ответов с полем start == true за период от одного пункта разбиения и до другого
+	Open       map[int64]int64   // количество ответов с полем start == true за период от одного пункта разбиения и до другого
-	Result     map[int64]int64  // количество ответов с полем result == true за период от одного пункта разбиения и до другого
+	Result     map[int64]int64   // количество ответов с полем result == true за период от одного пункта разбиения и до другого
-	AvTime     map[int64]uint64 // среднее время между ответом с полем result == true и start == true. в рамках сессии
+	AvTime     map[int64]uint64  // среднее время между ответом с полем result == true и start == true. в рамках сессии
-	Conversion map[int64]float64  // Result/Open за период от одного пункта разбиения и до другого
+	Conversion map[int64]float64 // Result/Open за период от одного пункта разбиения и до другого
 }
 
 func (r *StatisticsRepository) GetGeneralStatistics(ctx context.Context, req DeviceStatReq) (GeneralStatsResp, error) {
@@ -79,8 +78,8 @@ func (r *StatisticsRepository) GetGeneralStatistics(ctx context.Context, req Dev
 	}
 	// todo затестить запрос нужно, когда на один тру ответ приходится один тру старт апдейтнуть запрос
 	allStatistics, err := r.queries.GeneralStatistics(ctx, sqlcgen.GeneralStatisticsParams{
-		QuizID:  req.QuizId,
+		QuizID:        req.QuizId,
-		ToTimestamp: float64(req.From),
+		ToTimestamp:   float64(req.From),
 		ToTimestamp_2: float64(req.To),
 	})
 	if err != nil {
@@ -103,7 +102,7 @@ type QuestionsStatsResp struct {
 	// 0 - количество сессий с любым ответом кроме start == true /  количество сессий с ответом start == true
 	// 1 - количество сессий с result == false, но тип вопроса, на который ответ == result /  количество сессий с ответом start == true
 	// 2 - количество сессий с ответом result == true / количество сессий с ответом start == true
-	Funnel [3]float64
+	Funnel     [3]float64
 	FunnelData [4]float64
 	// ключ - заголовок  вопроса найденного по айдишнику вопроса в ответе result == true,
 	// значение - процент ответов с result == true и таким айдишником вопроса
@@ -115,10 +114,10 @@ type QuestionsStatsResp struct {
 
 func (r *StatisticsRepository) GetQuestionsStatistics(ctx context.Context, req DeviceStatReq) (QuestionsStatsResp, error) {
 	resp := QuestionsStatsResp{
-		Funnel:    [3]float64{},
+		Funnel:     [3]float64{},
-		FunnelData:    [4]float64{},
+		FunnelData: [4]float64{},
-		Results:   make(map[string]float64),
+		Results:    make(map[string]float64),
-		Questions: make(map[string]map[string]float64),
+		Questions:  make(map[string]map[string]float64),
 	}
 
 	queStatistics, err := r.queries.QuestionsStatistics(ctx, sqlcgen.QuestionsStatisticsParams{