Coverage for src/qdrant_loader/webhooks/handlers.py: 88%

1from __future__ import annotations 

2 

3import json 

4from typing import Any 

5 

6from qdrant_loader.core.worker.job_types import JobType 

7from qdrant_loader.utils.logging import LoggingConfig 

8from qdrant_loader.webhooks.queue_backend import ( 

9 FULL_SCAN, 

10 ChangeEvent, 

11 QueueBackendManager, 

12) 

13from qdrant_loader.webhooks.single_event_handler import parse_webhook_event 

14 

15logger = LoggingConfig.get_logger(__name__) 

16 

17# Single-event webhook support is Jira-only in v1.1. 

18SUPPORTED_SOURCE_TYPES = {"jira"} 

19 

20# Direct /ingest API supports all configured connector types. 

21INGEST_SUPPORTED_SOURCE_TYPES = { 

22 "jira", 

23 "confluence", 

24 "git", 

25 "publicdocs", 

26 "localfile", 

27} 

28 

29 

30def normalize_source_type(source_type: str) -> str: 

31 """Normalize and validate the source type from Jira webhook routes.""" 

32 normalized_source_type = source_type.strip().lower() 

33 if normalized_source_type not in SUPPORTED_SOURCE_TYPES: 

34 raise ValueError( 

35 f"Unsupported source type '{source_type}'. " 

36 f"Allowed values are: {', '.join(sorted(SUPPORTED_SOURCE_TYPES))}." 

37 ) 

38 return normalized_source_type 

39 

40 

41def normalize_ingest_source_type(source_type: str) -> str: 

42 """Normalize and validate the source type for direct /ingest requests.""" 

43 normalized_source_type = source_type.strip().lower() 

44 if normalized_source_type not in INGEST_SUPPORTED_SOURCE_TYPES: 

45 raise ValueError( 

46 f"Unsupported source type '{source_type}'. " 

47 f"Allowed values are: {', '.join(sorted(INGEST_SUPPORTED_SOURCE_TYPES))}." 

48 ) 

49 return normalized_source_type 

50 

51 

52def summarize_payload(payload: Any) -> str: 

53 """Create a short summary of webhook payload data for logging.""" 

54 try: 

55 summary = json.dumps(payload, default=str, ensure_ascii=False) 

56 except Exception: 

57 return "<non-serializable payload>" 

58 if len(summary) > 1000: 

59 return summary[:1000] + "..." 

60 return summary 

61 

62 

63async def enqueue_webhook_event( 

64 project_id: str | None, 

65 source_type: str, 

66 source: str, 

67 payload: Any, 

68 force: bool = False, 

69) -> dict[str, Any]: 

70 """Parse webhook payload and enqueue a durable change event. 

71 

72 Jira webhooks use SINGLE_UPSERT/SINGLE_DELETE when the payload is parseable. 

73 Falls back to FULL_SCAN when force=True or parsing fails. 

74 """ 

75 normalized_source_type = normalize_source_type(source_type) 

76 queue = QueueBackendManager.get_backend() 

77 

78 payload_summary = { 

79 "type": type(payload).__name__, 

80 "keys": sorted(payload.keys())[:20] if isinstance(payload, dict) else None, 

81 } 

82 logger.info( 

83 "Received webhook event", 

84 project_id=project_id, 

85 source_type=normalized_source_type, 

86 source=source, 

87 force=force, 

88 payload_meta=payload_summary, 

89 ) 

90 

91 if force: 

92 event = ChangeEvent( 

93 source=source, 

94 source_type=normalized_source_type, 

95 project_id=project_id, 

96 operation=FULL_SCAN, 

97 payload=payload, 

98 force=True, 

99 ) 

100 message_id = await queue.enqueue(event) 

101 return { 

102 "operation": FULL_SCAN, 

103 "message_id": message_id, 

104 "queued": True, 

105 } 

106 

107 change_event = await parse_webhook_event(normalized_source_type, source, payload) 

108 if change_event is not None: 

109 change_event.project_id = project_id 

110 message_id = await queue.enqueue(change_event) 

111 logger.info( 

112 "Enqueued single-event webhook", 

113 operation=change_event.operation, 

114 entity_id=change_event.entity_id, 

115 message_id=message_id, 

116 ) 

117 return { 

118 "operation": change_event.operation, 

119 "entity_id": change_event.entity_id, 

120 "message_id": message_id, 

121 "queued": True, 

122 } 

123 

124 logger.info( 

125 "Could not parse single-event; enqueueing full scan", 

126 source_type=normalized_source_type, 

127 source=source, 

128 ) 

129 event = ChangeEvent( 

130 source=source, 

131 source_type=normalized_source_type, 

132 project_id=project_id, 

133 operation=FULL_SCAN, 

134 payload=payload, 

135 force=False, 

136 ) 

137 message_id = await queue.enqueue(event) 

138 return { 

139 "operation": FULL_SCAN, 

140 "message_id": message_id, 

141 "queued": True, 

142 } 

143 

144 

145async def enqueue_ingest_request( 

146 project_id: str | None, 

147 source_type: str | None, 

148 source: str | None, 

149 force: bool = False, 

150) -> dict[str, Any]: 

151 """Enqueue a BULK_INGEST job processed by the serve worker pool. 

152 

153 Routes through IngestionJobHandler so the job reuses the already-warmed 

154 PipelineOrchestrator rather than spawning a standalone AsyncIngestionPipeline. 

155 source_type / source / project_id may all be None to mean "ingest everything". 

156 """ 

157 if source is not None and source_type is None: 

158 raise ValueError("source_type must be provided when source is specified.") 

159 

160 normalized_source_type = ( 

161 normalize_ingest_source_type(source_type) if source_type else None 

162 ) 

163 job_queue = QueueBackendManager.get_job_queue() 

164 

165 logger.info( 

166 "Received direct ingest request", 

167 project_id=project_id, 

168 source_type=normalized_source_type, 

169 source=source, 

170 force=force, 

171 ) 

172 

173 # source_lock serialises concurrent ingest requests for the same scope. 

174 # Use "*" as a wildcard token when a field is absent (meaning "all"). 

175 source_lock = ":".join( 

176 [ 

177 project_id or "*", 

178 normalized_source_type or "*", 

179 source or "*", 

180 ] 

181 ) 

182 payload: dict[str, Any] = { 

183 "project_id": project_id or "", 

184 "source_type": normalized_source_type or "", 

185 "source": source or "", 

186 "source_lock": source_lock, 

187 "force": force, 

188 } 

189 job = await job_queue.enqueue(JobType.BULK_INGEST, payload) 

190 return { 

191 "operation": JobType.BULK_INGEST, 

192 "message_id": str(job.id), 

193 "queued": True, 

194 }