headless_lms_chatbot/
azure_search_indexer.rs

1use crate::{
2    llm_utils::{azure_search_configuration, azure_search_request},
3    prelude::*,
4};
5use serde_json::json;
6
7const API_VERSION: &str = "2024-07-01";
8
9#[derive(Debug, Deserialize)]
10struct IndexerStatusResponse {
11    pub status: String,
12    #[serde(rename = "lastResult")]
13    pub last_result: Option<LastResult>,
14}
15
16#[derive(Debug, Deserialize)]
17struct LastResult {
18    pub status: String,
19    pub errors: Vec<IndexerError>,
20    pub warnings: Vec<IndexerWarning>,
21}
22
23#[derive(Debug, Deserialize)]
24struct IndexerError {
25    pub key: Option<String>,
26    pub name: Option<String>,
27    pub message: Option<String>,
28    pub details: Option<String>,
29    #[serde(rename = "documentationLink")]
30    pub documentation_link: Option<String>,
31}
32
33#[derive(Debug, Deserialize)]
34struct IndexerWarning {
35    key: Option<String>,
36    name: Option<String>,
37    message: Option<String>,
38    details: Option<String>,
39    #[serde(rename = "documentationLink")]
40    documentation_link: Option<String>,
41}
42
43pub async fn does_search_indexer_exist(
44    indexer_name: &str,
45    app_config: &ApplicationConfiguration,
46) -> ChatbotResult<bool> {
47    let search_config = azure_search_configuration(app_config)?;
48    let mut url = search_config.search_endpoint.clone();
49    url.set_path(&format!("indexers('{}')", indexer_name));
50    url.set_query(Some(&format!("api-version={}", API_VERSION)));
51
52    let response = azure_search_request(reqwest::Method::GET, url, search_config)
53        .send()
54        .await?;
55
56    if response.status().is_success() {
57        Ok(true)
58    } else if response.status() == 404 {
59        Ok(false)
60    } else {
61        let status = response.status();
62        let error_text = response.text().await?;
63        Err(chatbot_err!(
64            FailedAzureResponse,
65            format!(
66                "Error checking if index exists. Status: {}. Error: {}",
67                status, error_text
68            )
69        ))
70    }
71}
72
73pub async fn create_search_indexer(
74    indexer_name: &str,
75    data_source_name: &str,
76    skillset_name: &str,
77    target_index_name: &str,
78    app_config: &ApplicationConfiguration,
79) -> ChatbotResult<()> {
80    let search_config = azure_search_configuration(app_config)?;
81
82    let mut url = search_config.search_endpoint.clone();
83    url.set_path(&format!("indexers/{}", indexer_name));
84    url.set_query(Some(&format!("api-version={}", API_VERSION)));
85
86    let indexer_definition = json!({
87        "name": indexer_name,
88        "description": null,
89        "dataSourceName": data_source_name,
90        "skillsetName": skillset_name,
91        "targetIndexName": target_index_name,
92        "disabled": null,
93        "schedule": null,
94        "parameters": {
95            "batchSize": null,
96            "maxFailedItems": null,
97            "maxFailedItemsPerBatch": null,
98            "base64EncodeKeys": null,
99            "configuration": {
100                "dataToExtract": "contentAndMetadata"
101            }
102        },
103        "fieldMappings": [
104            {
105                "sourceFieldName": "metadata_storage_path",
106                "targetFieldName": "chunk_id",
107                "mappingFunction": { "name": "base64Encode" }
108            },
109        ],
110        "outputFieldMappings": [
111
112        ],
113        "encryptionKey": null
114    });
115
116    let response = azure_search_request(reqwest::Method::PUT, url, search_config)
117        .json(&indexer_definition)
118        .send()
119        .await?;
120
121    if response.status().is_success() {
122        Ok(())
123    } else {
124        let status = response.status();
125        let error_text = response.text().await?;
126        Err(chatbot_err!(
127            FailedAzureResponse,
128            format!(
129                "Error creating search indexer. Status: {}. Error: {}",
130                status, error_text
131            )
132        ))
133    }
134}
135
136pub async fn run_search_indexer_now(
137    indexer_name: &str,
138    app_config: &ApplicationConfiguration,
139) -> ChatbotResult<()> {
140    let search_config = azure_search_configuration(app_config)?;
141
142    let mut url = search_config.search_endpoint.clone();
143    url.set_path(&format!("indexers/{}/run", indexer_name));
144    url.set_query(Some(&format!("api-version={}", API_VERSION)));
145
146    let response = azure_search_request(reqwest::Method::POST, url, search_config)
147        .send()
148        .await?;
149
150    if response.status().is_success() {
151        Ok(())
152    } else {
153        let status = response.status();
154        let error_text = response.text().await?;
155        Err(chatbot_err!(
156            FailedAzureResponse,
157            format!(
158                "Error triggering search indexer. Status: {}. Error: {}",
159                status, error_text
160            )
161        ))
162    }
163}
164
165/// Checks if the search indexer exists, is not running, and the last execution is not in progress.
166/// Also prints any errors and warnings from the last execution in a nicely formatted manner.
167///
168/// # Arguments
169///
170/// * `indexer_name` - The name of the indexer to check.
171/// * `app_config` - The application configuration containing Azure settings.
172///
173/// # Returns
174///
175/// * `Ok(true)` if the indexer exists, is not running, and the last execution is not in progress.
176/// * `Ok(false)` if the indexer is running or the last execution is in progress.
177/// * An error if the indexer does not exist or if there's an issue fetching its status.
178pub async fn check_search_indexer_status(
179    indexer_name: &str,
180    app_config: &ApplicationConfiguration,
181) -> ChatbotResult<bool> {
182    let search_config = azure_search_configuration(app_config)?;
183
184    let mut url = search_config.search_endpoint.clone();
185    url.set_path(&format!("indexers('{}')/search.status", indexer_name));
186    url.set_query(Some(&format!("api-version={}", API_VERSION)));
187
188    let response = azure_search_request(reqwest::Method::GET, url, search_config)
189        .send()
190        .await?;
191
192    if response.status().is_success() {
193        let response_text = response.text().await?;
194        let indexer_status: IndexerStatusResponse = match serde_json::from_str(&response_text) {
195            Ok(status) => status,
196            Err(e) => {
197                error!("Failed to parse indexer status JSON: {}", e);
198                error!(
199                    "{}",
200                    serde_json::to_string_pretty(&response_text)
201                        .unwrap_or_else(|_| "Invalid JSON".to_string())
202                );
203                return Err(chatbot_err!(
204                    SerdeJson,
205                    "Failed to parse indexer status JSON.",
206                    e
207                ));
208            }
209        };
210
211        // If the indexer is not running, it is not healthy
212        let is_running = indexer_status.status.eq_ignore_ascii_case("running");
213
214        // If the last run is in progress, we cannot start a new run.
215        let last_result_in_progress = indexer_status
216            .last_result
217            .as_ref()
218            .is_some_and(|lr| lr.status.eq_ignore_ascii_case("inprogress"));
219
220        if !is_running {
221            info!("Indexer '{}' is not running normally.", indexer_name);
222        }
223
224        if last_result_in_progress {
225            warn!(
226                "Last execution of indexer '{}' is in progress.",
227                indexer_name
228            );
229        }
230
231        if let Some(last_result) = &indexer_status.last_result {
232            if !last_result.errors.is_empty() {
233                error!("Errors in the last execution:");
234                for error in &last_result.errors {
235                    error!(
236                        "  - **Key**: {}\n    **Name**: {}\n    **Message**: {}\n    **Details**: {}\n    **Documentation**: {}\n",
237                        error.key.as_deref().unwrap_or("N/A"),
238                        error.name.as_deref().unwrap_or("N/A"),
239                        error.message.as_deref().unwrap_or("N/A"),
240                        error.details.as_deref().unwrap_or("N/A"),
241                        error.documentation_link.as_deref().unwrap_or("N/A"),
242                    );
243                }
244            }
245
246            if !last_result.warnings.is_empty() {
247                warn!("Warnings in the last execution:");
248                for warning in &last_result.warnings {
249                    warn!(
250                        "  - **Key**: {}\n    **Name**: {}\n    **Message**: {}\n    **Details**: {}\n    **Documentation**: {}\n",
251                        warning.key.as_deref().unwrap_or("N/A"),
252                        warning.name.as_deref().unwrap_or("N/A"),
253                        warning.message.as_deref().unwrap_or("N/A"),
254                        warning.details.as_deref().unwrap_or("N/A"),
255                        warning.documentation_link.as_deref().unwrap_or("N/A"),
256                    );
257                }
258            }
259        } else {
260            warn!(
261                "No last result information available for indexer '{}'. Assuming the index is not ready yet.",
262                indexer_name
263            );
264            return Ok(false);
265        }
266
267        if is_running && !last_result_in_progress {
268            Ok(true)
269        } else {
270            Ok(false)
271        }
272    } else if response.status() == reqwest::StatusCode::NOT_FOUND {
273        error!("Indexer '{}' does not exist.", indexer_name);
274        Ok(false)
275    } else {
276        let status = response.status();
277        let error_text = response
278            .text()
279            .await
280            .unwrap_or_else(|_| "No error text".to_string());
281        error!(
282            "Error fetching indexer status. Status: {}. Error: {}",
283            status, error_text
284        );
285        Err(chatbot_err!(
286            FailedAzureResponse,
287            format!(
288                "Error fetching indexer status. Status: {}. Error: {}",
289                status, error_text
290            )
291        ))
292    }
293}
headless_lms_chatbot/azure_search_indexer.rs

headless_lms_chatbot/
azure_search_indexer.rs
