@Data
@NoArgsConstructor
public class Campaign implements Serializable {

	// 캠페인 아이디
	private Integer id;
	
	// 캠페인 명
	private String name;
	
	// 캠페인 설명
	private String description;
	
	// 캠페인 작성자, 등록자
	private String register;
	
	// 에이전트 타입
	private Agent type;
	
	// 발송 스케쥴
	private SendSchedule schedule;
	
	// 상태
	private State state = State.UNREGISTER;
	
	// 캠페인 최초 등록시간
	private LocalDateTime createdDate;
	
	// 캠페인 최종 수정시간
	private LocalDateTime modifiedDate;
	
	// 푸시 정보
	private Push push;
		
	//푸시 타이틀
	private String title;
		
	public Campaign(int id) {
		this.id = id;
	}
		
}

// 대상자 조회 프로세스 바인더
public interface TargetUserBinder {

	String TARGET_OUT = "target-out";
	String TARGET_IN = "target-in";
	
	@Output(TARGET_OUT)
	MessageChannel targetSearch();
	
	@Input(TARGET_IN)
	SubscribableChannel targetFilter();
	
}

// Push 메시지 생성 프로세스 바인더
public interface PushMessageBinder {

	String PROCESS_OUT = "process-out";
	String PROCESS_IN = "process-in";
	
	@Output(PROCESS_OUT)
	MessageChannel createMessage();
	
	@Input(PROCESS_IN)
	SubscribableChannel messagePush();
	
}

// 대상자 조회 후 대상자별로 데이터 설정 후 kafka 전송
@Slf4j
@EnableBinding(value= {TargetUserBinder.class, PushMessageBinder.class})
public class TargetUserProcess {
	
	private PushMessageBinder pushMessageBinder;
	
	@Autowired
	public TargetUserProcess(PushMessageBinder pushMessageBinder) {		
		this.pushMessageBinder = pushMessageBinder;
	}

	//바인더를 통한 데이터 전송
	@StreamListener(TargetUserBinder.TARGET_IN)
	public void targetUserSearch(@Payload TargetMeta targetMeta) {
		pushMessageBinder.createMessage().send(MessageBuilder.withPayload(targetMeta).build());
	}
}

// 대상자별 데이터를 기반으로 메시지 생성 후 전송, 결과 데이터 리턴
@Slf4j
@EnableBinding(value= {PushMessageBinder.class, ResultBinder.class})
public class PushMessageProcess {

	@StreamListener(PushMessageBinder.PROCESS_IN)
	@SendTo(ResultBinder.RESULT_OUT)  // 어노테이션 설정으로 데이터 전송
	public Summary messagePush(@Payload TargetMeta targetMeta) {
		return summary;
	}
}

@Slf4j
@EnableBinding(value= {ResultBinder.class, SummaryBinder.class})
public class SummaryProcess {
	
	@StreamListener(ResultBinder.RESULT_IN)
	@SendTo(SummaryBinder.SUMMARY_OUT)
	public KStream<Integer, Summary> process(KStream<Integer, Summary> input) {
		
		// 4. 요약 처리 
		return input
			.selectKey((k,v) -> v.getId())
			.groupByKey(Serialized.with(Serdes.Integer(), serde))
			.reduce((v1, v2) -> {
				Summary summary = new Summary(v1.getId(), v1.getEventMessageId(), v1.getCampaignId());
				summary.setSuccess(v1.getSuccess() + v2.getSuccess());
				summary.setError(v1.getError() + v2.getError());
				summary.setMissing(v1.getMissing() + v2.getMissing());
				summary.setFiltered(v1.getFiltered() + v2.getFiltered());
				return summary;
			}, Materialized.<Integer, Summary, KeyValueStore<Bytes, byte[]>>as(STORE_NAME).withKeySerde(Serdes.Integer()).withValueSerde(serde))
			.toStream().map((k, v) -> new KeyValue<>(k, v));
	}
	
	/*
	 * 5. 요약 수신
	 */
	@StreamListener(SummaryBinder.SUMMARY_IN)
	public void listener(@Payload Summary data) {
		// reduce 처리된 데이터에 대해 DB 업데이트 처리
	}
}