Function updateWshRunTEvent

pkg/telemetry/telemetry.go:201–236

aggregates wsh:run events per (cmd, haderror) key within the current 1-hour bucket

(ctx context.Context, tevent *telemetrydata.TEvent)

// aggregates wsh:run events per (cmd, haderror) key within the current 1-hour bucket
func updateWshRunTEvent(ctx context.Context, tevent *telemetrydata.TEvent) error {
	eventTs := time.Now().Truncate(time.Hour).Add(time.Hour)
	incomingCount := tevent.Props.WshCount
	if incomingCount <= 0 {
		incomingCount = 1
	}
	return wstore.WithTx(ctx, func(tx *wstore.TxWrap) error {
		uuidStr := tx.GetString(
			`SELECT uuid FROM db_tevent WHERE ts = ? AND event = ? AND json_extract(props, '$."wsh:cmd"') IS ?`,
			eventTs.UnixMilli(), WshRunEventName, tevent.Props.WshCmd,
		)
		if uuidStr != "" {
			var curProps telemetrydata.TEventProps
			rawProps := tx.GetString(`SELECT props FROM db_tevent WHERE uuid = ?`, uuidStr)
			if rawProps != "" {
				if err := json.Unmarshal([]byte(rawProps), &curProps); err != nil {
					log.Printf("error unmarshalling wsh:run props: %v\n", err)
				}
			}
			curCount := curProps.WshCount
			if curCount <= 0 {
				curCount = 1
			}
			curProps.WshCount = curCount + incomingCount
			curProps.WshErrorCount += tevent.Props.WshErrorCount
			tx.Exec(`UPDATE db_tevent SET props = ? WHERE uuid = ?`, dbutil.QuickJson(curProps), uuidStr)
		} else {
			newProps := tevent.Props
			newProps.WshCount = incomingCount
			tsLocal := utilfn.ConvertToWallClockPT(eventTs).Format(time.RFC3339)
			tx.Exec(`INSERT INTO db_tevent (uuid, ts, tslocal, event, props) VALUES (?, ?, ?, ?, ?)`,
				uuid.New().String(), eventTs.UnixMilli(), tsLocal, WshRunEventName, dbutil.QuickJson(newProps))
		}
		return nil
	})
}

func TruncateActivityTEventForShutdown(ctx context.Context) error {
	nowTs := time.Now()