Genarrative/server-rs/crates/module-runtime/src/agc_analytics.rs

//! 客户端观测事件校验，不推导项目所有权或奖励资格。
use serde_json::Value;
use shared_contracts::agc_analytics::*;
use uuid::Uuid;
const MAX_SAFE_INTEGER: u64 = 9_007_199_254_740_991;
trait EventDataRules {
    fn status(&self) -> Option<Status>;
    fn has_goal(&self) -> bool;
}
pub trait EventRules {
    fn data(&self) -> Result<EventData, &'static str>;
    fn validate(&self) -> Result<(), &'static str>;
}
impl EventDataRules for EventData {
    fn status(&self) -> Option<Status> {
        match self {
            Self::EditorFocusStart(_) | Self::EditorFocusEnd(_) => None,
            Self::AgentRunFailed(_) => Some(Status::Failed),
            _ => Some(Status::Success),
        }
    }

    fn has_goal(&self) -> bool {
        matches!(
            self,
            Self::CreativeTaskSubmit(_)
                | Self::AgentRunCompleted(_)
                | Self::AgentRunFailed(_)
                | Self::ProjectRevisionCreated(_)
                | Self::PreviewReady(_)
                | Self::ProjectSave(_)
        )
    }
}

impl EventRules for Event {
    fn data(&self) -> Result<EventData, &'static str> {
        if !self.properties.is_object() {
            return Err("invalid_properties");
        }
        // 可选字段不可得时省略；显式 null 仅用于合同指定的 nullable 字段。
        let optional = match self.event_name.as_str() {
            "project_create_success" => &["project_template_id"][..],
            "agent_run_completed" | "agent_run_failed" | "project_save" => &["revision_id"][..],
            "project_revision_created" => &["files_changed_count"][..],
            "preview_ready" => &["ready_duration_ms"][..],
            _ => &[][..],
        };
        if optional
            .iter()
            .any(|key| self.properties.get(key).is_some_and(Value::is_null))
        {
            return Err("invalid_optional_property");
        }
        serde_json::from_value(serde_json::json!({
            "event_name": self.event_name,
            "properties": self.properties,
        }))
        .map_err(|_| "invalid_properties")
    }

    fn validate(&self) -> Result<(), &'static str> {
        if self.schema_version != 1
            || !uuid(&self.event_id)
            || !uuid(&self.editor_session_id)
            || !id(&self.client_version)
            || !optional_id(&self.user_id)
            || !optional_id(&self.project_id)
            || !optional_id(&self.creative_task_id)
            || !optional_id(&self.agent_turn_id)
            || !valid_time(&self.event_time)
        {
            return Err("invalid_envelope");
        }
        let data = self.data()?;
        if self.status != data.status()
            || self.error_code.is_some() != matches!(data, EventData::AgentRunFailed(_))
        {
            return Err("invalid_status");
        }
        if data.has_goal() {
            if self.project_id.is_none() || self.creative_task_id != self.project_id {
                return Err("invalid_goal");
            }
        } else if self.creative_task_id.is_some() {
            return Err("unexpected_goal");
        }
        let is_run = matches!(
            data,
            EventData::AgentRunCompleted(_) | EventData::AgentRunFailed(_)
        );
        if is_run {
            if !self.agent_run_id.as_deref().is_some_and(uuid) {
                return Err("invalid_run");
            }
        } else if self.agent_run_id.is_some() || self.agent_turn_id.is_some() {
            return Err("unexpected_run");
        }
        let editor = self.source == Source::Editor;
        let agent = matches!(self.source, Source::Direct | Source::DesignAgent);
        let valid = match data {
            EventData::EditorSessionStart(p) => editor && p.first_project_id == self.project_id,
            EventData::EditorSessionEnd(p) => {
                editor && p.last_project_id == self.project_id && safe(p.session_duration_ms)
            }
            EventData::EditorFocusStart(p) => {
                editor && uuid(&p.focus_interval_id) && p.active_project_id == self.project_id
            }
            EventData::EditorFocusEnd(p) => {
                editor
                    && uuid(&p.focus_interval_id)
                    && p.active_project_id == self.project_id
                    && safe(p.focus_duration_ms)
            }
            EventData::ProjectCreateSuccess(p) => {
                editor && self.project_id.is_some() && optional_id(&p.project_template_id)
            }
            EventData::ProjectOpen(_) => editor && self.project_id.is_some(),
            EventData::CreativeTaskSubmit(_) => agent,
            EventData::AgentRunCompleted(p) | EventData::AgentRunFailed(p) => {
                agent
                    && ((self.source == Source::Direct) == (p.agent_type == AgentType::GameAgent))
                    && ((self.status == Some(Status::Failed))
                        == (p.end_reason == RunEndReason::Failed))
                    && safe(p.duration_ms)
                    && safe(Some(p.retry_index))
                    && optional_id(&p.revision_id)
            }
            EventData::ProjectRevisionCreated(p) => {
                id(&p.revision_id)
                    && safe(p.files_changed_count)
                    && match p.revision_source {
                        RevisionSource::Agent => agent,
                        RevisionSource::AssetCanvas => self.source == Source::AssetCanvas,
                        RevisionSource::ResourceEditor => self.source == Source::ResourceEditor,
                        RevisionSource::UiEditor => self.source == Source::UiEditor,
                        RevisionSource::ManualEdit => self.source == Source::Manual,
                        RevisionSource::SystemProjection => self.source == Source::System,
                    }
            }
            EventData::PreviewReady(p) => id(&p.preview_version) && safe(p.ready_duration_ms),
            EventData::ProjectSave(p) => optional_id(&p.revision_id),
        };
        if valid {
            Ok(())
        } else {
            Err("invalid_event_fields")
        }
    }
}

fn id(value: &str) -> bool {
    !value.trim().is_empty() && value.len() <= 256 && !value.chars().any(char::is_control)
}
fn optional_id(value: &Option<String>) -> bool {
    value.as_deref().is_none_or(id)
}
fn uuid(value: &str) -> bool {
    Uuid::parse_str(value).is_ok_and(|v| v.get_version_num() == 4 && v.to_string() == value)
}
fn safe(value: Option<u64>) -> bool {
    value.is_none_or(|v| v <= MAX_SAFE_INTEGER)
}
fn valid_time(value: &str) -> bool {
    value.len() == 24
        && value.ends_with("Z")
        && value.as_bytes()[10] == b'T'
        && value.as_bytes()[19] == b'.'
        && time::OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339)
            .is_ok()
}

pub fn validate_agc_analytics_batch(batch: &AgcAnalyticsBatch) -> Result<(), &'static str> {
    if batch.schema_version != 1
        || batch.destination_origin.len() > 2048
        || !uuid(&batch.batch_id)
        || !id(&batch.user_id)
        || batch.events.is_empty()
        || batch.events.len() > 500
    {
        return Err("invalid_batch");
    }
    let origin = url::Url::parse(&batch.destination_origin).map_err(|_| "invalid_origin")?;
    if !matches!(origin.scheme(), "https" | "http")
        || origin.host_str().is_none()
        || !origin.username().is_empty()
        || origin.password().is_some()
        || origin.origin().ascii_serialization() != batch.destination_origin
    {
        return Err("invalid_origin");
    }
    let mut ids = std::collections::HashSet::new();
    let mut bytes = 0usize;
    for event in &batch.events {
        event.validate()?;
        if event.user_id.as_deref() != Some(batch.user_id.as_str()) {
            return Err("identity_mismatch");
        }
        if !ids.insert(&event.event_id) {
            return Err("duplicate_event_id");
        }
        bytes += serde_json::to_vec(event)
            .map_err(|_| "invalid_event")?
            .len()
            + 1;
    }
    if bytes > 1024 * 1024 {
        return Err("events_too_large");
    }
    Ok(())
}

#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct AgcTrackingCursor {
    pub snapshot: i64,
    pub event_time: i64,
    pub event_id: String,
    pub filter_key: String,
}
/// 游标绑定查询条件；分页大小不参与条件，刷新时不带游标。
pub fn agc_tracking_filter_key(
    query: &shared_contracts::admin::AdminAgcTrackingEventListQuery,
) -> String {
    let mut filters = query.clone();
    filters.cursor = None;
    filters.limit = None;
    serde_json::to_string(&filters).expect("query serializes")
}
pub fn agc_time_micros(value: &str) -> Result<i64, &'static str> {
    let time = time::OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339)
        .map_err(|_| "invalid_agc_query")?;
    i64::try_from(time.unix_timestamp_nanos() / 1000).map_err(|_| "invalid_agc_query")
}
pub fn validate_agc_tracking_query(
    query: &shared_contracts::admin::AdminAgcTrackingEventListQuery,
) -> Result<Option<AgcTrackingCursor>, &'static str> {
    for value in [
        &query.user_id,
        &query.project_id,
        &query.creative_task_id,
        &query.agent_run_id,
        &query.event_name,
        &query.client_version,
    ]
    .into_iter()
    .flatten()
    {
        if !id(value) {
            return Err("invalid_agc_query");
        }
    }
    let start = query
        .start_time
        .as_deref()
        .map(agc_time_micros)
        .transpose()?;
    let end = query.end_time.as_deref().map(agc_time_micros).transpose()?;
    if start.zip(end).is_some_and(|(a, b)| a >= b) {
        return Err("invalid_agc_query");
    }
    if let Some(raw) = &query.cursor {
        if raw.len() > 8192 {
            return Err("invalid_agc_cursor");
        }
        let cursor: AgcTrackingCursor =
            serde_json::from_str(raw).map_err(|_| "invalid_agc_cursor")?;
        if cursor.filter_key != agc_tracking_filter_key(query) || !uuid(&cursor.event_id) {
            return Err("invalid_agc_cursor");
        }
        Ok(Some(cursor))
    } else {
        Ok(None)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;
    const ID: &str = "790a1275-a0a0-405c-8bfd-287201bef10a";
    fn event(name: &str, properties: Value) -> Event {
        serde_json::from_value(json!({
            "schema_version": 1, "event_id": ID, "event_name": name,
            "event_time": "2026-09-21T12:00:00.000Z", "user_id": "user-a",
            "editor_session_id": ID, "project_id": null, "creative_task_id": null,
            "agent_run_id": null, "agent_turn_id": null, "status": "success",
            "error_code": null, "source": "editor", "client_version": "1.0.0", "properties": properties
        })).unwrap()
    }
    fn batch(event: Event) -> AgcAnalyticsBatch {
        AgcAnalyticsBatch {
            schema_version: 1,
            batch_id: ID.into(),
            destination_origin: "https://example.com".into(),
            user_id: "user-a".into(),
            events: vec![event],
        }
    }
    #[test]
    fn agc_accepts_twelve_existing_event_contracts() {
        let cases = [
            (
                "editor_session_start",
                json!({"entry_source":"direct_launch","first_project_id":null}),
            ),
            (
                "editor_session_end",
                json!({"end_reason":"user_exit","session_duration_ms":null,"last_project_id":null}),
            ),
            (
                "editor_focus_start",
                json!({"focus_interval_id":ID,"focus_reason":"window_focus","active_project_id":null}),
            ),
            (
                "editor_focus_end",
                json!({"focus_interval_id":ID,"blur_reason":"window_blur","focus_duration_ms":1,"active_project_id":null}),
            ),
            (
                "project_create_success",
                json!({"creation_source":"home_game"}),
            ),
            (
                "project_open",
                json!({"open_source":"recent","is_first_open":null}),
            ),
            ("creative_task_submit", json!({})),
            (
                "agent_run_completed",
                json!({"agent_type":"game_agent","run_source":"user_submit","duration_ms":1,"retry_index":0,"output_change_detected":null,"end_reason":"waiting_for_approval"}),
            ),
            (
                "agent_run_failed",
                json!({"agent_type":"game_agent","run_source":"user_submit","duration_ms":1,"retry_index":0,"output_change_detected":true,"end_reason":"failed"}),
            ),
            (
                "project_revision_created",
                json!({"revision_id":"design:session:phase","revision_source":"agent","change_kind":"design_document"}),
            ),
            (
                "preview_ready",
                json!({"preview_source":"user","preview_version":"version1"}),
            ),
            ("project_save", json!({"save_source":"checkpoint"})),
        ];
        for (index, (name, properties)) in cases.into_iter().enumerate() {
            let mut e = event(name, properties);
            if index == 2 || index == 3 {
                e.status = None;
            }
            if index >= 4 {
                e.project_id = Some("project-a".into());
            }
            if index >= 6 {
                e.creative_task_id = e.project_id.clone();
                e.source = Source::Direct;
            }
            if index == 7 || index == 8 {
                e.agent_run_id = Some(ID.into());
            }
            if index == 8 {
                e.status = Some(Status::Failed);
                e.error_code = Some(ErrorCode::ProviderTimeout);
            }
            assert_eq!(validate_agc_analytics_batch(&batch(e)), Ok(()), "{name}");
        }
    }
    #[test]
    fn agc_rejects_identity_duplicate_and_unknown_content() {
        let e = event(
            "editor_session_start",
            json!({"entry_source":"direct_launch","first_project_id":null}),
        );
        let mut b = batch(e.clone());
        b.user_id = "other-user".into();
        assert_eq!(validate_agc_analytics_batch(&b), Err("identity_mismatch"));
        let mut b = batch(e.clone());
        b.events.push(e.clone());
        assert_eq!(validate_agc_analytics_batch(&b), Err("duplicate_event_id"));
        let mut b = batch(e.clone());
        b.events[0].properties["prompt"] = json!("should never be accepted");
        assert_eq!(validate_agc_analytics_batch(&b), Err("invalid_properties"));
        b.events[0].properties = json!(["direct_launch", null]);
        assert_eq!(validate_agc_analytics_batch(&b), Err("invalid_properties"));
        let mut raw = serde_json::to_value(e).unwrap();
        raw["file_path"] = json!("secret");
        assert!(serde_json::from_value::<Event>(raw).is_err());
    }
    #[test]
    fn agc_requires_explicit_nullable_fields_and_no_optional_nulls() {
        let mut raw =
            serde_json::to_value(event("project_save", json!({"save_source":"manual"}))).unwrap();
        raw.as_object_mut().unwrap().remove("agent_turn_id");
        assert!(serde_json::from_value::<Event>(raw).is_err());
        let mut e = event(
            "project_save",
            json!({"save_source":"manual","revision_id":null}),
        );
        e.project_id = Some("p".into());
        e.creative_task_id = e.project_id.clone();
        assert_eq!(e.validate(), Err("invalid_optional_property"));
    }
    #[test]
    fn agc_query_cursor_is_bound_to_filters_and_rejects_invalid_time_range() {
        let mut query = shared_contracts::admin::AdminAgcTrackingEventListQuery::default();
        let cursor = AgcTrackingCursor {
            snapshot: 100,
            // 客户端时钟可能领先；快照限制入库时间，不能限制发生时间。
            event_time: 110,
            event_id: ID.into(),
            filter_key: agc_tracking_filter_key(&query),
        };
        query.cursor = Some(serde_json::to_string(&cursor).unwrap());
        assert!(validate_agc_tracking_query(&query).is_ok());
        query.user_id = Some("changed-user".into());
        assert_eq!(
            validate_agc_tracking_query(&query).unwrap_err(),
            "invalid_agc_cursor"
        );
        query.cursor = None;
        query.start_time = Some("2026-09-22T00:00:00Z".into());
        query.end_time = Some("2026-09-21T00:00:00Z".into());
        assert_eq!(
            validate_agc_tracking_query(&query).unwrap_err(),
            "invalid_agc_query"
        );
    }
}