GuanM/certimate
1
0
Fork 0
mirror of https://github.com/usual2970/certimate.git synced 2025-10-04 21:44:54 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge branch 'feat/new-workflow' of github.com:fudiwei/certimate into next

Browse Source
This commit is contained in:
Yoan.liu
2025-02-12 09:42:00 +08:00
parent 1f6b33f4f6 94408e8a9f
commit 138e08e985
97 changed files with 2288 additions and 1138 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
internal/applicant/acme_user.go
View File

@@ -1,6 +1,7 @@
package applicant
import (
"context"
"crypto"
"crypto/ecdsa"
"crypto/elliptic"
@@ -110,14 +111,11 @@ func registerAcmeUser(client *lego.Client, sslProviderConfig *acmeSSLProviderCon
Kid: sslProviderConfig.Config.GoogleTrustServices.EabKid,
HmacEncoded: sslProviderConfig.Config.GoogleTrustServices.EabHmacKey,
})
case sslProviderLetsEncrypt, sslProviderLetsEncryptStaging:
reg, err = client.Registration.Register(registration.RegisterOptions{TermsOfServiceAgreed: true})
default:
err = fmt.Errorf("unsupported ssl provider: %s", sslProviderConfig.Provider)
}
if err != nil {
return nil, err
}
@@ -129,7 +127,12 @@ func registerAcmeUser(client *lego.Client, sslProviderConfig *acmeSSLProviderCon
return resp.Resource, nil
}
if err := repo.Save(sslProviderConfig.Provider, user.GetEmail(), user.getPrivateKeyPEM(), reg); err != nil {
if _, err := repo.Save(context.Background(), &domain.AcmeAccount{
CA: sslProviderConfig.Provider,
Email: user.GetEmail(),
Key: user.getPrivateKeyPEM(),
Resource: reg,
}); err != nil {
return nil, fmt.Errorf("failed to save registration: %w", err)
}

30
internal/applicant/applicant.go
View File

@@ -26,6 +26,7 @@ type ApplyCertResult struct {
CertificateFullChain string
IssuerCertificate string
PrivateKey string
ACMEAccountUrl string
ACMECertUrl string
ACMECertStableUrl string
CSR string
@@ -46,8 +47,7 @@ type applicantOptions struct {
DnsPropagationTimeout int32
DnsTTL int32
DisableFollowCNAME bool
DisableARI bool
SkipBeforeExpiryDays int32
ReplacedARIAcctId string
ReplacedARICertId string
}
@@ -67,8 +67,6 @@ func NewWithApplyNode(node *domain.WorkflowNode) (Applicant, error) {
DnsPropagationTimeout: nodeConfig.DnsPropagationTimeout,
DnsTTL: nodeConfig.DnsTTL,
DisableFollowCNAME: nodeConfig.DisableFollowCNAME,
DisableARI: nodeConfig.DisableARI,
SkipBeforeExpiryDays: nodeConfig.SkipBeforeExpiryDays,
}
accessRepo := repository.NewAccessRepository()
@@ -95,6 +93,7 @@ func NewWithApplyNode(node *domain.WorkflowNode) (Applicant, error) {
lastCertX509, _ := certcrypto.ParsePEMCertificate([]byte(lastCertificate.Certificate))
if lastCertX509 != nil {
replacedARICertId, _ := certificate.MakeARICertID(lastCertX509)
options.ReplacedARIAcctId = lastCertificate.ACMEAccountUrl
options.ReplacedARICertId = replacedARICertId
}
}
@@ -141,7 +140,7 @@ func apply(challengeProvider challenge.Provider, options *applicantOptions) (*Ap
// Create an ACME client config
config := lego.NewConfig(acmeUser)
config.CADirURL = sslProviderUrls[sslProviderConfig.Provider]
config.Certificate.KeyType = parseKeyAlgorithm(options.KeyAlgorithm)
config.Certificate.KeyType = parseKeyAlgorithm(domain.CertificateKeyAlgorithmType(options.KeyAlgorithm))
// Create an ACME client
client, err := lego.NewClient(config)
@@ -171,7 +170,7 @@ func apply(challengeProvider challenge.Provider, options *applicantOptions) (*Ap
Domains: options.Domains,
Bundle: true,
}
if !options.DisableARI {
if options.ReplacedARICertId != "" && options.ReplacedARIAcctId != acmeUser.Registration.URI {
certRequest.ReplacesCertID = options.ReplacedARICertId
}
certResource, err := client.Certificate.Obtain(certRequest)
@@ -183,29 +182,30 @@ func apply(challengeProvider challenge.Provider, options *applicantOptions) (*Ap
CertificateFullChain: strings.TrimSpace(string(certResource.Certificate)),
IssuerCertificate: strings.TrimSpace(string(certResource.IssuerCertificate)),
PrivateKey: strings.TrimSpace(string(certResource.PrivateKey)),
ACMEAccountUrl: acmeUser.Registration.URI,
ACMECertUrl: certResource.CertURL,
ACMECertStableUrl: certResource.CertStableURL,
CSR: strings.TrimSpace(string(certResource.CSR)),
}, nil
}
func parseKeyAlgorithm(algo string) certcrypto.KeyType {
func parseKeyAlgorithm(algo domain.CertificateKeyAlgorithmType) certcrypto.KeyType {
switch algo {
case "RSA2048":
case domain.CertificateKeyAlgorithmTypeRSA2048:
return certcrypto.RSA2048
case "RSA3072":
case domain.CertificateKeyAlgorithmTypeRSA3072:
return certcrypto.RSA3072
case "RSA4096":
case domain.CertificateKeyAlgorithmTypeRSA4096:
return certcrypto.RSA4096
case "RSA8192":
case domain.CertificateKeyAlgorithmTypeRSA8192:
return certcrypto.RSA8192
case "EC256":
case domain.CertificateKeyAlgorithmTypeEC256:
return certcrypto.EC256
case "EC384":
case domain.CertificateKeyAlgorithmTypeEC384:
return certcrypto.EC384
default:
return certcrypto.RSA2048
}
return certcrypto.RSA2048
}
// TODO: 暂时使用代理模式以兼容之前版本代码，后续重新实现此处逻辑

68
internal/applicant/providers.go
View File

@@ -35,8 +35,8 @@ func createApplicant(options *applicantOptions) (challenge.Provider, error) {
case domain.ApplyDNSProviderTypeACMEHttpReq:
{
access := domain.AccessConfigForACMEHttpReq{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
applicant, err := providerACMEHttpReq.NewChallengeProvider(&providerACMEHttpReq.ACMEHttpReqApplicantConfig{
@@ -52,8 +52,8 @@ func createApplicant(options *applicantOptions) (challenge.Provider, error) {
case domain.ApplyDNSProviderTypeAliyun, domain.ApplyDNSProviderTypeAliyunDNS:
{
access := domain.AccessConfigForAliyun{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
applicant, err := providerAliyun.NewChallengeProvider(&providerAliyun.AliyunApplicantConfig{
@@ -68,8 +68,8 @@ func createApplicant(options *applicantOptions) (challenge.Provider, error) {
case domain.ApplyDNSProviderTypeAWS, domain.ApplyDNSProviderTypeAWSRoute53:
{
access := domain.AccessConfigForAWS{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
applicant, err := providerAWSRoute53.NewChallengeProvider(&providerAWSRoute53.AWSRoute53ApplicantConfig{
@@ -86,8 +86,8 @@ func createApplicant(options *applicantOptions) (challenge.Provider, error) {
case domain.ApplyDNSProviderTypeAzureDNS:
{
access := domain.AccessConfigForAzure{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
applicant, err := providerAzureDNS.NewChallengeProvider(&providerAzureDNS.AzureDNSApplicantConfig{
@@ -104,8 +104,8 @@ func createApplicant(options *applicantOptions) (challenge.Provider, error) {
case domain.ApplyDNSProviderTypeCloudflare:
{
access := domain.AccessConfigForCloudflare{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
applicant, err := providerCloudflare.NewChallengeProvider(&providerCloudflare.CloudflareApplicantConfig{
@@ -119,8 +119,8 @@ func createApplicant(options *applicantOptions) (challenge.Provider, error) {
case domain.ApplyDNSProviderTypeClouDNS:
{
access := domain.AccessConfigForClouDNS{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
applicant, err := providerClouDNS.NewChallengeProvider(&providerClouDNS.ClouDNSApplicantConfig{
@@ -135,8 +135,8 @@ func createApplicant(options *applicantOptions) (challenge.Provider, error) {
case domain.ApplyDNSProviderTypeGname:
{
access := domain.AccessConfigForGname{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
applicant, err := providerGname.NewChallengeProvider(&providerGname.GnameApplicantConfig{
@@ -151,8 +151,8 @@ func createApplicant(options *applicantOptions) (challenge.Provider, error) {
case domain.ApplyDNSProviderTypeGoDaddy:
{
access := domain.AccessConfigForGoDaddy{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
applicant, err := providerGoDaddy.NewChallengeProvider(&providerGoDaddy.GoDaddyApplicantConfig{
@@ -167,8 +167,8 @@ func createApplicant(options *applicantOptions) (challenge.Provider, error) {
case domain.ApplyDNSProviderTypeHuaweiCloud, domain.ApplyDNSProviderTypeHuaweiCloudDNS:
{
access := domain.AccessConfigForHuaweiCloud{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
applicant, err := providerHuaweiCloud.NewChallengeProvider(&providerHuaweiCloud.HuaweiCloudApplicantConfig{
@@ -184,8 +184,8 @@ func createApplicant(options *applicantOptions) (challenge.Provider, error) {
case domain.ApplyDNSProviderTypeNameDotCom:
{
access := domain.AccessConfigForNameDotCom{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
applicant, err := providerNameDotCom.NewChallengeProvider(&providerNameDotCom.NameDotComApplicantConfig{
@@ -200,8 +200,8 @@ func createApplicant(options *applicantOptions) (challenge.Provider, error) {
case domain.ApplyDNSProviderTypeNameSilo:
{
access := domain.AccessConfigForNameSilo{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
applicant, err := providerNameSilo.NewChallengeProvider(&providerNameSilo.NameSiloApplicantConfig{
@@ -215,8 +215,8 @@ func createApplicant(options *applicantOptions) (challenge.Provider, error) {
case domain.ApplyDNSProviderTypeNS1:
{
access := domain.AccessConfigForNS1{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
applicant, err := providerNS1.NewChallengeProvider(&providerNS1.NS1ApplicantConfig{
@@ -230,8 +230,8 @@ func createApplicant(options *applicantOptions) (challenge.Provider, error) {
case domain.ApplyDNSProviderTypePowerDNS:
{
access := domain.AccessConfigForPowerDNS{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
applicant, err := providerPowerDNS.NewChallengeProvider(&providerPowerDNS.PowerDNSApplicantConfig{
@@ -246,8 +246,8 @@ func createApplicant(options *applicantOptions) (challenge.Provider, error) {
case domain.ApplyDNSProviderTypeRainYun:
{
access := domain.AccessConfigForRainYun{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
applicant, err := providerRainYun.NewChallengeProvider(&providerRainYun.RainYunApplicantConfig{
@@ -261,8 +261,8 @@ func createApplicant(options *applicantOptions) (challenge.Provider, error) {
case domain.ApplyDNSProviderTypeTencentCloud, domain.ApplyDNSProviderTypeTencentCloudDNS:
{
access := domain.AccessConfigForTencentCloud{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
applicant, err := providerTencentCloud.NewChallengeProvider(&providerTencentCloud.TencentCloudApplicantConfig{
@@ -277,8 +277,8 @@ func createApplicant(options *applicantOptions) (challenge.Provider, error) {
case domain.ApplyDNSProviderTypeVolcEngine, domain.ApplyDNSProviderTypeVolcEngineDNS:
{
access := domain.AccessConfigForVolcEngine{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
applicant, err := providerVolcEngine.NewChallengeProvider(&providerVolcEngine.VolcEngineApplicantConfig{
@@ -293,8 +293,8 @@ func createApplicant(options *applicantOptions) (challenge.Provider, error) {
case domain.ApplyDNSProviderTypeWestcn:
{
access := domain.AccessConfigForWestcn{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
applicant, err := providerWestcn.NewChallengeProvider(&providerWestcn.WestcnApplicantConfig{

56
internal/certificate/service.go
View File

@@ -5,7 +5,7 @@ import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"strconv"
"strings"
"time"
@@ -30,18 +30,18 @@ type certificateRepository interface {
}
type CertificateService struct {
repo certificateRepository
certRepo certificateRepository
}
func NewCertificateService(repo certificateRepository) *CertificateService {
func NewCertificateService(certRepo certificateRepository) *CertificateService {
return &CertificateService{
repo: repo,
certRepo: certRepo,
}
}
func (s *CertificateService) InitSchedule(ctx context.Context) error {
app.GetScheduler().MustAdd("certificateExpireSoonNotify", "0 0 * * *", func() {
certs, err := s.repo.ListExpireSoon(context.Background())
certs, err := s.certRepo.ListExpireSoon(context.Background())
if err != nil {
app.GetLogger().Error("failed to get certificates which expire soon", "err", err)
return
@@ -59,8 +59,8 @@ func (s *CertificateService) InitSchedule(ctx context.Context) error {
return nil
}
func (s *CertificateService) ArchiveFile(ctx context.Context, req *dtos.CertificateArchiveFileReq) ([]byte, error) {
certificate, err := s.repo.GetById(ctx, req.CertificateId)
func (s *CertificateService) ArchiveFile(ctx context.Context, req *dtos.CertificateArchiveFileReq) (*dtos.CertificateArchiveFileResp, error) {
certificate, err := s.certRepo.GetById(ctx, req.CertificateId)
if err != nil {
return nil, err
}
@@ -69,6 +69,10 @@ func (s *CertificateService) ArchiveFile(ctx context.Context, req *dtos.Certific
zipWriter := zip.NewWriter(&buf)
defer zipWriter.Close()
resp := &dtos.CertificateArchiveFileResp{
FileFormat: "zip",
}
switch strings.ToUpper(req.Format) {
case "", "PEM":
{
@@ -97,7 +101,8 @@ func (s *CertificateService) ArchiveFile(ctx context.Context, req *dtos.Certific
return nil, err
}
return buf.Bytes(), nil
resp.FileBytes = buf.Bytes()
return resp, nil
}
case "PFX":
@@ -134,7 +139,8 @@ func (s *CertificateService) ArchiveFile(ctx context.Context, req *dtos.Certific
return nil, err
}
return buf.Bytes(), nil
resp.FileBytes = buf.Bytes()
return resp, nil
}
case "JKS":
@@ -171,7 +177,8 @@ func (s *CertificateService) ArchiveFile(ctx context.Context, req *dtos.Certific
return nil, err
}
return buf.Bytes(), nil
resp.FileBytes = buf.Bytes()
return resp, nil
}
default:
@@ -180,25 +187,30 @@ func (s *CertificateService) ArchiveFile(ctx context.Context, req *dtos.Certific
}
func (s *CertificateService) ValidateCertificate(ctx context.Context, req *dtos.CertificateValidateCertificateReq) (*dtos.CertificateValidateCertificateResp, error) {
info, err := certs.ParseCertificateFromPEM(req.Certificate)
certX509, err := certs.ParseCertificateFromPEM(req.Certificate)
if err != nil {
return nil, err
} else if time.Now().After(certX509.NotAfter) {
return nil, fmt.Errorf("certificate has expired at %s", certX509.NotAfter.UTC().Format(time.RFC3339))
}
return &dtos.CertificateValidateCertificateResp{
IsValid: true,
Domains: strings.Join(certX509.DNSNames, ";"),
}, nil
}
func (s *CertificateService) ValidatePrivateKey(ctx context.Context, req *dtos.CertificateValidatePrivateKeyReq) (*dtos.CertificateValidatePrivateKeyResp, error) {
_, err := certcrypto.ParsePEMPrivateKey([]byte(req.PrivateKey))
if err != nil {
return nil, err
}
if time.Now().After(info.NotAfter) {
return nil, errors.New("证书已过期")
}
return &dtos.CertificateValidateCertificateResp{
Domains: strings.Join(info.DNSNames, ";"),
return &dtos.CertificateValidatePrivateKeyResp{
IsValid: true,
}, nil
}
func (s *CertificateService) ValidatePrivateKey(ctx context.Context, req *dtos.CertificateValidatePrivateKeyReq) error {
_, err := certcrypto.ParsePEMPrivateKey([]byte(req.PrivateKey))
return err
}
func buildExpireSoonNotification(certificates []*domain.Certificate) *struct {
Subject string
Message string

56
internal/deployer/providers.go
View File

@@ -54,8 +54,8 @@ func createDeployer(options *deployerOptions) (deployer.Deployer, logger.Logger,
case domain.DeployProviderTypeAliyunALB, domain.DeployProviderTypeAliyunCDN, domain.DeployProviderTypeAliyunCLB, domain.DeployProviderTypeAliyunDCDN, domain.DeployProviderTypeAliyunLive, domain.DeployProviderTypeAliyunNLB, domain.DeployProviderTypeAliyunOSS, domain.DeployProviderTypeAliyunWAF:
{
access := domain.AccessConfigForAliyun{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
switch options.Provider {
@@ -146,8 +146,8 @@ func createDeployer(options *deployerOptions) (deployer.Deployer, logger.Logger,
case domain.DeployProviderTypeAWSCloudFront:
{
access := domain.AccessConfigForAWS{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
switch options.Provider {
@@ -168,8 +168,8 @@ func createDeployer(options *deployerOptions) (deployer.Deployer, logger.Logger,
case domain.DeployProviderTypeBaiduCloudCDN:
{
access := domain.AccessConfigForBaiduCloud{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
switch options.Provider {
@@ -189,8 +189,8 @@ func createDeployer(options *deployerOptions) (deployer.Deployer, logger.Logger,
case domain.DeployProviderTypeBytePlusCDN:
{
access := domain.AccessConfigForBytePlus{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
switch options.Provider {
@@ -210,8 +210,8 @@ func createDeployer(options *deployerOptions) (deployer.Deployer, logger.Logger,
case domain.DeployProviderTypeDogeCloudCDN:
{
access := domain.AccessConfigForDogeCloud{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
deployer, err := providerDogeCDN.NewWithLogger(&providerDogeCDN.DogeCloudCDNDeployerConfig{
@@ -225,8 +225,8 @@ func createDeployer(options *deployerOptions) (deployer.Deployer, logger.Logger,
case domain.DeployProviderTypeEdgioApplications:
{
access := domain.AccessConfigForEdgio{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
deployer, err := providerEdgioApplications.NewWithLogger(&providerEdgioApplications.EdgioApplicationsDeployerConfig{
@@ -240,8 +240,8 @@ func createDeployer(options *deployerOptions) (deployer.Deployer, logger.Logger,
case domain.DeployProviderTypeHuaweiCloudCDN, domain.DeployProviderTypeHuaweiCloudELB:
{
access := domain.AccessConfigForHuaweiCloud{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
switch options.Provider {
@@ -291,8 +291,8 @@ func createDeployer(options *deployerOptions) (deployer.Deployer, logger.Logger,
case domain.DeployProviderTypeKubernetesSecret:
{
access := domain.AccessConfigForKubernetes{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
deployer, err := providerK8sSecret.NewWithLogger(&providerK8sSecret.K8sSecretDeployerConfig{
@@ -309,8 +309,8 @@ func createDeployer(options *deployerOptions) (deployer.Deployer, logger.Logger,
case domain.DeployProviderTypeQiniuCDN, domain.DeployProviderTypeQiniuPili:
{
access := domain.AccessConfigForQiniu{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
switch options.Provider {
@@ -339,8 +339,8 @@ func createDeployer(options *deployerOptions) (deployer.Deployer, logger.Logger,
case domain.DeployProviderTypeSSH:
{
access := domain.AccessConfigForSSH{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
deployer, err := providerSSH.NewWithLogger(&providerSSH.SshDeployerConfig{
@@ -367,8 +367,8 @@ func createDeployer(options *deployerOptions) (deployer.Deployer, logger.Logger,
case domain.DeployProviderTypeTencentCloudCDN, domain.DeployProviderTypeTencentCloudCLB, domain.DeployProviderTypeTencentCloudCOS, domain.DeployProviderTypeTencentCloudCSS, domain.DeployProviderTypeTencentCloudECDN, domain.DeployProviderTypeTencentCloudEO:
{
access := domain.AccessConfigForTencentCloud{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
switch options.Provider {
@@ -435,8 +435,8 @@ func createDeployer(options *deployerOptions) (deployer.Deployer, logger.Logger,
case domain.DeployProviderTypeUCloudUCDN, domain.DeployProviderTypeUCloudUS3:
{
access := domain.AccessConfigForUCloud{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
switch options.Provider {
@@ -468,8 +468,8 @@ func createDeployer(options *deployerOptions) (deployer.Deployer, logger.Logger,
case domain.DeployProviderTypeVolcEngineCDN, domain.DeployProviderTypeVolcEngineCLB, domain.DeployProviderTypeVolcEngineDCDN, domain.DeployProviderTypeVolcEngineLive, domain.DeployProviderTypeVolcEngineTOS:
{
access := domain.AccessConfigForVolcEngine{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
switch options.Provider {
@@ -525,8 +525,8 @@ func createDeployer(options *deployerOptions) (deployer.Deployer, logger.Logger,
case domain.DeployProviderTypeWebhook:
{
access := domain.AccessConfigForWebhook{}
if err := maps.Decode(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to decode provider access config: %w", err)
if err := maps.Populate(options.ProviderAccessConfig, &access); err != nil {
return nil, nil, fmt.Errorf("failed to populate provider access config: %w", err)
}
deployer, err := providerWebhook.NewWithLogger(&providerWebhook.WebhookDeployerConfig{

93
internal/domain/certificate.go
View File

@@ -1,24 +1,77 @@
package domain
import "time"
import (
"crypto/x509"
"strings"
"time"
"github.com/usual2970/certimate/internal/pkg/utils/certs"
)
const CollectionNameCertificate = "certificate"
type Certificate struct {
Meta
Source CertificateSourceType `json:"source" db:"source"`
SubjectAltNames string `json:"subjectAltNames" db:"subjectAltNames"`
Certificate string `json:"certificate" db:"certificate"`
PrivateKey string `json:"privateKey" db:"privateKey"`
IssuerCertificate string `json:"issuerCertificate" db:"issuerCertificate"`
EffectAt time.Time `json:"effectAt" db:"effectAt"`
ExpireAt time.Time `json:"expireAt" db:"expireAt"`
ACMECertUrl string `json:"acmeCertUrl" db:"acmeCertUrl"`
ACMECertStableUrl string `json:"acmeCertStableUrl" db:"acmeCertStableUrl"`
WorkflowId string `json:"workflowId" db:"workflowId"`
WorkflowNodeId string `json:"workflowNodeId" db:"workflowNodeId"`
WorkflowOutputId string `json:"workflowOutputId" db:"workflowOutputId"`
DeletedAt *time.Time `json:"deleted" db:"deleted"`
Source CertificateSourceType `json:"source" db:"source"`
SubjectAltNames string `json:"subjectAltNames" db:"subjectAltNames"`
SerialNumber string `json:"serialNumber" db:"serialNumber"`
Certificate string `json:"certificate" db:"certificate"`
PrivateKey string `json:"privateKey" db:"privateKey"`
Issuer string `json:"issuer" db:"issuer"`
IssuerCertificate string `json:"issuerCertificate" db:"issuerCertificate"`
KeyAlgorithm CertificateKeyAlgorithmType `json:"keyAlgorithm" db:"keyAlgorithm"`
EffectAt time.Time `json:"effectAt" db:"effectAt"`
ExpireAt time.Time `json:"expireAt" db:"expireAt"`
ACMEAccountUrl string `json:"acmeAccountUrl" db:"acmeAccountUrl"`
ACMECertUrl string `json:"acmeCertUrl" db:"acmeCertUrl"`
ACMECertStableUrl string `json:"acmeCertStableUrl" db:"acmeCertStableUrl"`
WorkflowId string `json:"workflowId" db:"workflowId"`
WorkflowNodeId string `json:"workflowNodeId" db:"workflowNodeId"`
WorkflowRunId string `json:"workflowRunId" db:"workflowRunId"`
WorkflowOutputId string `json:"workflowOutputId" db:"workflowOutputId"`
DeletedAt *time.Time `json:"deleted" db:"deleted"`
}
func (c *Certificate) PopulateFromX509(certX509 *x509.Certificate) *Certificate {
c.SubjectAltNames = strings.Join(certX509.DNSNames, ";")
c.SerialNumber = strings.ToUpper(certX509.SerialNumber.Text(16))
c.Issuer = strings.Join(certX509.Issuer.Organization, ";")
c.EffectAt = certX509.NotBefore
c.ExpireAt = certX509.NotAfter
switch certX509.SignatureAlgorithm {
case x509.SHA256WithRSA, x509.SHA256WithRSAPSS:
c.KeyAlgorithm = CertificateKeyAlgorithmTypeRSA2048
case x509.SHA384WithRSA, x509.SHA384WithRSAPSS:
c.KeyAlgorithm = CertificateKeyAlgorithmTypeRSA3072
case x509.SHA512WithRSA, x509.SHA512WithRSAPSS:
c.KeyAlgorithm = CertificateKeyAlgorithmTypeRSA4096
case x509.ECDSAWithSHA256:
c.KeyAlgorithm = CertificateKeyAlgorithmTypeEC256
case x509.ECDSAWithSHA384:
c.KeyAlgorithm = CertificateKeyAlgorithmTypeEC384
case x509.ECDSAWithSHA512:
c.KeyAlgorithm = CertificateKeyAlgorithmTypeEC512
default:
c.KeyAlgorithm = CertificateKeyAlgorithmType("")
}
return c
}
func (c *Certificate) PopulateFromPEM(certPEM, privkeyPEM string) *Certificate {
c.Certificate = certPEM
c.PrivateKey = privkeyPEM
_, issuerCertPEM, _ := certs.ExtractCertificatesFromPEM(certPEM)
c.IssuerCertificate = issuerCertPEM
certX509, _ := certs.ParseCertificateFromPEM(certPEM)
if certX509 != nil {
c.PopulateFromX509(certX509)
}
return c
}
type CertificateSourceType string
@@ -27,3 +80,15 @@ const (
CertificateSourceTypeWorkflow = CertificateSourceType("workflow")
CertificateSourceTypeUpload = CertificateSourceType("upload")
)
type CertificateKeyAlgorithmType string
const (
CertificateKeyAlgorithmTypeRSA2048 = CertificateKeyAlgorithmType("RSA2048")
CertificateKeyAlgorithmTypeRSA3072 = CertificateKeyAlgorithmType("RSA3072")
CertificateKeyAlgorithmTypeRSA4096 = CertificateKeyAlgorithmType("RSA4096")
CertificateKeyAlgorithmTypeRSA8192 = CertificateKeyAlgorithmType("RSA8192")
CertificateKeyAlgorithmTypeEC256 = CertificateKeyAlgorithmType("EC256")
CertificateKeyAlgorithmTypeEC384 = CertificateKeyAlgorithmType("EC384")
CertificateKeyAlgorithmTypeEC512 = CertificateKeyAlgorithmType("EC512")
)

16
internal/domain/dtos/certificate.go
View File

@@ -5,22 +5,24 @@ type CertificateArchiveFileReq struct {
Format string `json:"format"`
}
type CertificateArchiveFileResp struct {
FileBytes []byte `json:"fileBytes"`
FileFormat string `json:"fileFormat"`
}
type CertificateValidateCertificateReq struct {
Certificate string `json:"certificate"`
}
type CertificateValidateCertificateResp struct {
Domains string `json:"domains"`
IsValid bool `json:"isValid"`
Domains string `json:"domains,omitempty"`
}
type CertificateValidatePrivateKeyReq struct {
PrivateKey string `json:"privateKey"`
}
type CertificateUploadReq struct {
WorkflowId string `json:"workflowId"`
WorkflowNodeId string `json:"workflowNodeId"`
CertificateId string `json:"certificateId"`
Certificate string `json:"certificate"`
PrivateKey string `json:"privateKey"`
type CertificateValidatePrivateKeyResp struct {
IsValid bool `json:"isValid"`
}

2
internal/domain/dtos/workflow.go
View File

@@ -4,7 +4,7 @@ import "github.com/usual2970/certimate/internal/domain"
type WorkflowStartRunReq struct {
WorkflowId string `json:"-"`
Trigger domain.WorkflowTriggerType `json:"trigger"`
RunTrigger domain.WorkflowTriggerType `json:"trigger"`
}
type WorkflowCancelRunReq struct {

5
internal/domain/workflow.go
View File

@@ -55,8 +55,8 @@ type WorkflowNode struct {
Inputs []WorkflowNodeIO `json:"inputs"`
Outputs []WorkflowNodeIO `json:"outputs"`
Next *WorkflowNode `json:"next"`
Branches []WorkflowNode `json:"branches"`
Next *WorkflowNode `json:"next,omitempty"`
Branches []WorkflowNode `json:"branches,omitempty"`
Validated bool `json:"validated"`
}
@@ -64,6 +64,7 @@ type WorkflowNode struct {
type WorkflowNodeConfigForApply struct {
Domains string `json:"domains"` // 域名列表，以半角逗号分隔
ContactEmail string `json:"contactEmail"` // 联系邮箱
ChallengeType string `json:"challengeType"` // TODO: 验证方式。目前仅支持 dns-01
Provider string `json:"provider"` // DNS 提供商
ProviderAccessId string `json:"providerAccessId"` // DNS 提供商授权记录 ID
ProviderConfig map[string]any `json:"providerConfig"` // DNS 提供商额外配置

1
internal/domain/workflow_output.go
View File

@@ -5,6 +5,7 @@ const CollectionNameWorkflowOutput = "workflow_output"
type WorkflowOutput struct {
Meta
WorkflowId string `json:"workflowId" db:"workflow"`
RunId string `json:"runId" db:"runId"`
NodeId string `json:"nodeId" db:"nodeId"`
Node *WorkflowNode `json:"node" db:"node"`
Outputs []WorkflowNodeIO `json:"outputs" db:"outputs"`

21
internal/domain/workflow_run.go
View File

@@ -31,17 +31,26 @@ const (
type WorkflowRunLog struct {
NodeId string `json:"nodeId"`
NodeName string `json:"nodeName"`
Records []WorkflowRunLogRecord `json:"records"`
Error string `json:"error"`
Outputs []WorkflowRunLogOutput `json:"outputs"`
}
type WorkflowRunLogOutput struct {
Time string `json:"time"`
Title string `json:"title"`
Content string `json:"content"`
Error string `json:"error"`
type WorkflowRunLogRecord struct {
Time string `json:"time"`
Level WorkflowRunLogLevel `json:"level"`
Content string `json:"content"`
Error string `json:"error"`
}
type WorkflowRunLogLevel string
const (
WorkflowRunLogLevelDebug WorkflowRunLogLevel = "DEBUG"
WorkflowRunLogLevelInfo WorkflowRunLogLevel = "INFO"
WorkflowRunLogLevelWarn WorkflowRunLogLevel = "WARN"
WorkflowRunLogLevelError WorkflowRunLogLevel = "ERROR"
)
type WorkflowRunLogs []WorkflowRunLog
func (r WorkflowRunLogs) ErrorString() string {

2
internal/pkg/core/applicant/acme-dns-01/lego-providers/gname/internal/lego.go
View File

@@ -173,8 +173,6 @@ func (d *DNSProvider) addOrUpdateDNSRecord(domain, subDomain, value string) erro
_, err := d.client.ModifyDomainResolution(request)
return err
}
return nil
}
func (d *DNSProvider) removeDNSRecord(domain, subDomain, value string) error {

36
internal/pkg/core/deployer/providers/edgio-applications/edgio_applications.go
View File

@@ -2,13 +2,13 @@
import (
"context"
"encoding/pem"
"errors"
xerrors "github.com/pkg/errors"
"github.com/usual2970/certimate/internal/pkg/core/deployer"
"github.com/usual2970/certimate/internal/pkg/core/logger"
"github.com/usual2970/certimate/internal/pkg/utils/certs"
edgsdk "github.com/usual2970/certimate/internal/pkg/vendors/edgio-sdk/applications/v7"
edgsdkDtos "github.com/usual2970/certimate/internal/pkg/vendors/edgio-sdk/applications/v7/dtos"
)
@@ -57,7 +57,10 @@ func NewWithLogger(config *EdgioApplicationsDeployerConfig, logger logger.Logger
func (d *EdgioApplicationsDeployer) Deploy(ctx context.Context, certPem string, privkeyPem string) (*deployer.DeployResult, error) {
// 提取 Edgio 所需的服务端证书和中间证书内容
privateCertPem, intermediateCertPem := extractCertChains(certPem)
privateCertPem, intermediateCertPem, err := certs.ExtractCertificatesFromPEM(certPem)
if err != nil {
return nil, err
}
// 上传 TLS 证书
// REF: https://docs.edg.io/rest_api/#tag/tls-certs/operation/postConfigV01TlsCerts
@@ -81,32 +84,3 @@ func createSdkClient(clientId, clientSecret string) (*edgsdk.EdgioClient, error)
client := edgsdk.NewEdgioClient(clientId, clientSecret, "", "")
return client, nil
}
func extractCertChains(certPem string) (primaryCertPem string, intermediateCertPem string) {
pemBlocks := make([]*pem.Block, 0)
pemData := []byte(certPem)
for {
block, rest := pem.Decode(pemData)
if block == nil {
break
}
pemBlocks = append(pemBlocks, block)
pemData = rest
}
primaryCertPem = ""
intermediateCertPem = ""
if len(pemBlocks) > 0 {
primaryCertPem = string(pem.EncodeToMemory(pemBlocks[0]))
}
if len(pemBlocks) > 1 {
for i := 1; i < len(pemBlocks); i++ {
intermediateCertPem += string(pem.EncodeToMemory(pemBlocks[i]))
}
}
return primaryCertPem, intermediateCertPem
}

6
internal/pkg/core/deployer/providers/qiniu-cdn/qiniu_cdn.go
View File

@@ -78,7 +78,7 @@ func (d *QiniuCDNDeployer) Deploy(ctx context.Context, certPem string, privkeyPe
// 获取域名信息
// REF: https://developer.qiniu.com/fusion/4246/the-domain-name
getDomainInfoResp, err := d.sdkClient.GetDomainInfo(domain)
getDomainInfoResp, err := d.sdkClient.GetDomainInfo(context.TODO(), domain)
if err != nil {
return nil, xerrors.Wrap(err, "failed to execute sdk request 'cdn.GetDomainInfo'")
}
@@ -88,14 +88,14 @@ func (d *QiniuCDNDeployer) Deploy(ctx context.Context, certPem string, privkeyPe
// 判断域名是否已启用 HTTPS。如果已启用，修改域名证书；否则，启用 HTTPS
// REF: https://developer.qiniu.com/fusion/4246/the-domain-name
if getDomainInfoResp.Https != nil && getDomainInfoResp.Https.CertID != "" {
modifyDomainHttpsConfResp, err := d.sdkClient.ModifyDomainHttpsConf(domain, upres.CertId, getDomainInfoResp.Https.ForceHttps, getDomainInfoResp.Https.Http2Enable)
modifyDomainHttpsConfResp, err := d.sdkClient.ModifyDomainHttpsConf(context.TODO(), domain, upres.CertId, getDomainInfoResp.Https.ForceHttps, getDomainInfoResp.Https.Http2Enable)
if err != nil {
return nil, xerrors.Wrap(err, "failed to execute sdk request 'cdn.ModifyDomainHttpsConf'")
}
d.logger.Logt("已修改域名证书", modifyDomainHttpsConfResp)
} else {
enableDomainHttpsResp, err := d.sdkClient.EnableDomainHttps(domain, upres.CertId, true, true)
enableDomainHttpsResp, err := d.sdkClient.EnableDomainHttps(context.TODO(), domain, upres.CertId, true, true)
if err != nil {
return nil, xerrors.Wrap(err, "failed to execute sdk request 'cdn.EnableDomainHttps'")
}

2
internal/pkg/core/uploader/providers/qiniu-sslcert/qiniu_sslcert.go
View File

@@ -60,7 +60,7 @@ func (u *QiniuSSLCertUploader) Upload(ctx context.Context, certPem string, privk
// 上传新证书
// REF: https://developer.qiniu.com/fusion/8593/interface-related-certificate
uploadSslCertResp, err := u.sdkClient.UploadSslCert(certName, certX509.Subject.CommonName, certPem, privkeyPem)
uploadSslCertResp, err := u.sdkClient.UploadSslCert(context.TODO(), certName, certX509.Subject.CommonName, certPem, privkeyPem)
if err != nil {
return nil, xerrors.Wrap(err, "failed to execute sdk request 'cdn.UploadSslCert'")
}

48
internal/pkg/utils/certs/extractor.go Normal file
View File

@@ -0,0 +1,48 @@
package certs
import (
"encoding/pem"
"errors"
)
// 从 PEM 编码的证书字符串解析并提取服务器证书和中间证书。
//
// 入参:
// - certPem: 证书 PEM 内容。
//
// 出参:
// - serverCertPem: 服务器证书的 PEM 内容。
// - interCertPem: 中间证书的 PEM 内容。
// - err: 错误。
func ExtractCertificatesFromPEM(certPem string) (serverCertPem string, interCertPem string, err error) {
pemBlocks := make([]*pem.Block, 0)
pemData := []byte(certPem)
for {
block, rest := pem.Decode(pemData)
if block == nil || block.Type != "CERTIFICATE" {
break
}
pemBlocks = append(pemBlocks, block)
pemData = rest
}
serverCertPem = ""
interCertPem = ""
if len(pemBlocks) == 0 {
return "", "", errors.New("failed to decode PEM block")
}
if len(pemBlocks) > 0 {
serverCertPem = string(pem.EncodeToMemory(pemBlocks[0]))
}
if len(pemBlocks) > 1 {
for i := 1; i < len(pemBlocks); i++ {
interCertPem += string(pem.EncodeToMemory(pemBlocks[i]))
}
}
return serverCertPem, interCertPem, nil
}

1
internal/pkg/utils/certs/parser.go
View File

@@ -13,6 +13,7 @@ import (
)
// 从 PEM 编码的证书字符串解析并返回一个 x509.Certificate 对象。
// PEM 内容可能是包含多张证书的证书链，但只返回第一个证书（即服务器证书）。
//
// 入参:
// - certPem: 证书 PEM 内容。

11
internal/pkg/utils/maps/maps.go
View File

@@ -183,7 +183,7 @@ func GetValueOrDefaultAsBool(dict map[string]any, key string, defaultValue bool)
return defaultValue
}
// 将字典解码为指定类型的结构体。
// 将字典填充到指定类型的结构体。
// 与 [json.Unmarshal] 类似，但传入的是一个 [map[string]interface{}] 对象而非 JSON 格式的字符串。
//
// 入参：
@@ -191,8 +191,8 @@ func GetValueOrDefaultAsBool(dict map[string]any, key string, defaultValue bool)
// - output: 结构体指针。
//
// 出参：
// - 错误信息。如果解码失败，则返回错误信息。
func Decode(dict map[string]any, output any) error {
// - 错误信息。如果填充失败，则返回错误信息。
func Populate(dict map[string]any, output any) error {
config := &mapstructure.DecoderConfig{
Metadata: nil,
Result: output,
@@ -207,3 +207,8 @@ func Decode(dict map[string]any, output any) error {
return decoder.Decode(dict)
}
// Deprecated: Use [Populate] instead.
func Decode(dict map[string]any, output any) error {
return Populate(dict, output)
}

2
internal/pkg/vendors/gname-sdk/client.go vendored
View File

@@ -150,7 +150,7 @@ func (c *GnameClient) sendRequestWithResult(path string, params map[string]any,
if err := json.Unmarshal(resp.Body(), &jsonResp); err != nil {
return fmt.Errorf("failed to parse response: %w", err)
}
if err := maps.Decode(jsonResp, &result); err != nil {
if err := maps.Populate(jsonResp, &result); err != nil {
return fmt.Errorf("failed to parse response: %w", err)
}

29
internal/pkg/vendors/qiniu-sdk/auth.go vendored Normal file
View File

@@ -0,0 +1,29 @@
package qiniusdk
import (
"net/http"
"github.com/qiniu/go-sdk/v7/auth"
)
type transport struct {
http.RoundTripper
mac *auth.Credentials
}
func newTransport(mac *auth.Credentials, tr http.RoundTripper) *transport {
if tr == nil {
tr = http.DefaultTransport
}
return &transport{tr, mac}
}
func (t *transport) RoundTrip(req *http.Request) (resp *http.Response, err error) {
token, err := t.mac.SignRequestV2(req)
if err != nil {
return
}
req.Header.Set("Authorization", "Qiniu "+token)
return t.RoundTripper.RoundTrip(req)
}

122
internal/pkg/vendors/qiniu-sdk/client.go vendored
View File

@@ -1,48 +1,40 @@
package qiniusdk
import (
"bytes"
"encoding/json"
"context"
"fmt"
"io"
"net/http"
"strings"
"github.com/qiniu/go-sdk/v7/auth"
"github.com/qiniu/go-sdk/v7/client"
)
const qiniuHost = "https://api.qiniu.com"
type Client struct {
mac *auth.Credentials
client *client.Client
}
func NewClient(mac *auth.Credentials) *Client {
if mac == nil {
mac = auth.Default()
}
return &Client{mac: mac}
client := client.DefaultClient
client.Transport = newTransport(mac, nil)
return &Client{client: &client}
}
func (c *Client) GetDomainInfo(domain string) (*GetDomainInfoResponse, error) {
respBytes, err := c.sendReq(http.MethodGet, fmt.Sprintf("domain/%s", domain), nil)
if err != nil {
func (c *Client) GetDomainInfo(ctx context.Context, domain string) (*GetDomainInfoResponse, error) {
resp := new(GetDomainInfoResponse)
if err := c.client.Call(ctx, resp, http.MethodGet, c.urlf("domain/%s", domain), nil); err != nil {
return nil, err
}
resp := &GetDomainInfoResponse{}
err = json.Unmarshal(respBytes, resp)
if err != nil {
return nil, err
}
if resp.Code != nil && *resp.Code != 0 && *resp.Code != 200 {
return nil, fmt.Errorf("code: %d, error: %s", *resp.Code, *resp.Error)
}
return resp, nil
}
func (c *Client) ModifyDomainHttpsConf(domain, certId string, forceHttps, http2Enable bool) (*ModifyDomainHttpsConfResponse, error) {
func (c *Client) ModifyDomainHttpsConf(ctx context.Context, domain string, certId string, forceHttps bool, http2Enable bool) (*ModifyDomainHttpsConfResponse, error) {
req := &ModifyDomainHttpsConfRequest{
DomainInfoHttpsData: DomainInfoHttpsData{
CertID: certId,
@@ -50,30 +42,14 @@ func (c *Client) ModifyDomainHttpsConf(domain, certId string, forceHttps, http2E
Http2Enable: http2Enable,
},
}
reqBytes, err := json.Marshal(req)
if err != nil {
resp := new(ModifyDomainHttpsConfResponse)
if err := c.client.CallWithJson(ctx, resp, http.MethodPut, c.urlf("domain/%s/httpsconf", domain), nil, req); err != nil {
return nil, err
}
respBytes, err := c.sendReq(http.MethodPut, fmt.Sprintf("domain/%s/httpsconf", domain), bytes.NewReader(reqBytes))
if err != nil {
return nil, err
}
resp := &ModifyDomainHttpsConfResponse{}
err = json.Unmarshal(respBytes, resp)
if err != nil {
return nil, err
}
if resp.Code != nil && *resp.Code != 0 && *resp.Code != 200 {
return nil, fmt.Errorf("code: %d, error: %s", *resp.Code, *resp.Error)
}
return resp, nil
}
func (c *Client) EnableDomainHttps(domain, certId string, forceHttps, http2Enable bool) (*EnableDomainHttpsResponse, error) {
func (c *Client) EnableDomainHttps(ctx context.Context, domain string, certId string, forceHttps bool, http2Enable bool) (*EnableDomainHttpsResponse, error) {
req := &EnableDomainHttpsRequest{
DomainInfoHttpsData: DomainInfoHttpsData{
CertID: certId,
@@ -81,83 +57,29 @@ func (c *Client) EnableDomainHttps(domain, certId string, forceHttps, http2Enabl
Http2Enable: http2Enable,
},
}
reqBytes, err := json.Marshal(req)
if err != nil {
resp := new(EnableDomainHttpsResponse)
if err := c.client.CallWithJson(ctx, resp, http.MethodPut, c.urlf("domain/%s/sslize", domain), nil, req); err != nil {
return nil, err
}
respBytes, err := c.sendReq(http.MethodPut, fmt.Sprintf("domain/%s/sslize", domain), bytes.NewReader(reqBytes))
if err != nil {
return nil, err
}
resp := &EnableDomainHttpsResponse{}
err = json.Unmarshal(respBytes, resp)
if err != nil {
return nil, err
}
if resp.Code != nil && *resp.Code != 0 && *resp.Code != 200 {
return nil, fmt.Errorf("code: %d, error: %s", *resp.Code, *resp.Error)
}
return resp, nil
}
func (c *Client) UploadSslCert(name, commonName, certificate, privateKey string) (*UploadSslCertResponse, error) {
func (c *Client) UploadSslCert(ctx context.Context, name string, commonName string, certificate string, privateKey string) (*UploadSslCertResponse, error) {
req := &UploadSslCertRequest{
Name: name,
CommonName: commonName,
Certificate: certificate,
PrivateKey: privateKey,
}
reqBytes, err := json.Marshal(req)
if err != nil {
resp := new(UploadSslCertResponse)
if err := c.client.CallWithJson(ctx, resp, http.MethodPost, c.urlf("sslcert"), nil, req); err != nil {
return nil, err
}
respBytes, err := c.sendReq(http.MethodPost, "sslcert", bytes.NewReader(reqBytes))
if err != nil {
return nil, err
}
resp := &UploadSslCertResponse{}
err = json.Unmarshal(respBytes, resp)
if err != nil {
return nil, err
}
if resp.Code != nil && *resp.Code != 0 && *resp.Code != 200 {
return nil, fmt.Errorf("qiniu api error, code: %d, error: %s", *resp.Code, *resp.Error)
}
return resp, nil
}
func (c *Client) sendReq(method string, path string, body io.Reader) ([]byte, error) {
func (c *Client) urlf(pathf string, pathargs ...any) string {
path := fmt.Sprintf(pathf, pathargs...)
path = strings.TrimPrefix(path, "/")
req, err := http.NewRequest(method, fmt.Sprintf("%s/%s", qiniuHost, path), body)
if err != nil {
return nil, err
}
req.Header.Add("Content-Type", "application/json")
if err := c.mac.AddToken(auth.TokenQBox, req); err != nil {
return nil, err
}
client := http.Client{}
resp, err := client.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
r, err := io.ReadAll(resp.Body)
if err != nil {
return nil, err
}
return r, nil
return qiniuHost + "/" + path
}

2
internal/pkg/vendors/qiniu-sdk/models.go vendored
View File

@@ -13,7 +13,7 @@ type UploadSslCertRequest struct {
}
type UploadSslCertResponse struct {
*BaseResponse
BaseResponse
CertID string `json:"certID"`
}

38
internal/repository/acme_account.go
View File

@@ -1,6 +1,9 @@
package repository
import (
"context"
"database/sql"
"errors"
"fmt"
"github.com/go-acme/lego/v4/registration"
@@ -48,18 +51,37 @@ func (r *AcmeAccountRepository) GetByCAAndEmail(ca, email string) (*domain.AcmeA
return r.castRecordToModel(record)
}
func (r *AcmeAccountRepository) Save(ca, email, key string, resource *registration.Resource) error {
func (r *AcmeAccountRepository) Save(ctx context.Context, acmeAccount *domain.AcmeAccount) (*domain.AcmeAccount, error) {
collection, err := app.GetApp().FindCollectionByNameOrId(domain.CollectionNameAcmeAccount)
if err != nil {
return err
return acmeAccount, err
}
record := core.NewRecord(collection)
record.Set("ca", ca)
record.Set("email", email)
record.Set("key", key)
record.Set("resource", resource)
return app.GetApp().Save(record)
var record *core.Record
if acmeAccount.Id == "" {
record = core.NewRecord(collection)
} else {
record, err = app.GetApp().FindRecordById(collection, acmeAccount.Id)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return acmeAccount, domain.ErrRecordNotFound
}
return acmeAccount, err
}
}
record.Set("ca", acmeAccount.CA)
record.Set("email", acmeAccount.Email)
record.Set("key", acmeAccount.Key)
record.Set("resource", acmeAccount.Resource)
if err := app.GetApp().Save(record); err != nil {
return acmeAccount, err
}
acmeAccount.Id = record.Id
acmeAccount.CreatedAt = record.GetDateTime("created").Time()
acmeAccount.UpdatedAt = record.GetDateTime("updated").Time()
return acmeAccount, nil
}
func (r *AcmeAccountRepository) castRecordToModel(record *core.Record) (*domain.AcmeAccount, error) {

51
internal/repository/certificate.go
View File

@@ -79,6 +79,52 @@ func (r *CertificateRepository) GetByWorkflowNodeId(ctx context.Context, workflo
return r.castRecordToModel(records[0])
}
func (r *CertificateRepository) Save(ctx context.Context, certificate *domain.Certificate) (*domain.Certificate, error) {
collection, err := app.GetApp().FindCollectionByNameOrId(domain.CollectionNameCertificate)
if err != nil {
return certificate, err
}
var record *core.Record
if certificate.Id == "" {
record = core.NewRecord(collection)
} else {
record, err = app.GetApp().FindRecordById(collection, certificate.Id)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return certificate, domain.ErrRecordNotFound
}
return certificate, err
}
}
record.Set("source", string(certificate.Source))
record.Set("subjectAltNames", certificate.SubjectAltNames)
record.Set("serialNumber", certificate.SerialNumber)
record.Set("certificate", certificate.Certificate)
record.Set("privateKey", certificate.PrivateKey)
record.Set("issuer", certificate.Issuer)
record.Set("issuerCertificate", certificate.IssuerCertificate)
record.Set("keyAlgorithm", string(certificate.KeyAlgorithm))
record.Set("effectAt", certificate.EffectAt)
record.Set("expireAt", certificate.ExpireAt)
record.Set("acmeAccountUrl", certificate.ACMEAccountUrl)
record.Set("acmeCertUrl", certificate.ACMECertUrl)
record.Set("acmeCertStableUrl", certificate.ACMECertStableUrl)
record.Set("workflowId", certificate.WorkflowId)
record.Set("workflowRunId", certificate.WorkflowRunId)
record.Set("workflowNodeId", certificate.WorkflowNodeId)
record.Set("workflowOutputId", certificate.WorkflowOutputId)
if err := app.GetApp().Save(record); err != nil {
return certificate, err
}
certificate.Id = record.Id
certificate.CreatedAt = record.GetDateTime("created").Time()
certificate.UpdatedAt = record.GetDateTime("updated").Time()
return certificate, nil
}
func (r *CertificateRepository) castRecordToModel(record *core.Record) (*domain.Certificate, error) {
if record == nil {
return nil, fmt.Errorf("record is nil")
@@ -92,14 +138,19 @@ func (r *CertificateRepository) castRecordToModel(record *core.Record) (*domain.
},
Source: domain.CertificateSourceType(record.GetString("source")),
SubjectAltNames: record.GetString("subjectAltNames"),
SerialNumber: record.GetString("serialNumber"),
Certificate: record.GetString("certificate"),
PrivateKey: record.GetString("privateKey"),
Issuer: record.GetString("issuer"),
IssuerCertificate: record.GetString("issuerCertificate"),
KeyAlgorithm: domain.CertificateKeyAlgorithmType(record.GetString("keyAlgorithm")),
EffectAt: record.GetDateTime("effectAt").Time(),
ExpireAt: record.GetDateTime("expireAt").Time(),
ACMEAccountUrl: record.GetString("acmeAccountUrl"),
ACMECertUrl: record.GetString("acmeCertUrl"),
ACMECertStableUrl: record.GetString("acmeCertStableUrl"),
WorkflowId: record.GetString("workflowId"),
WorkflowRunId: record.GetString("workflowRunId"),
WorkflowNodeId: record.GetString("workflowNodeId"),
WorkflowOutputId: record.GetString("workflowOutputId"),
}

64
internal/repository/workflow.go
View File

@@ -24,7 +24,7 @@ func (r *WorkflowRepository) ListEnabledAuto(ctx context.Context) ([]*domain.Wor
"enabled={:enabled} && trigger={:trigger}",
"-created",
0, 0,
dbx.Params{"enabled": true, "trigger": domain.WorkflowTriggerTypeAuto},
dbx.Params{"enabled": true, "trigger": string(domain.WorkflowTriggerTypeAuto)},
)
if err != nil {
return nil, err
@@ -65,7 +65,7 @@ func (r *WorkflowRepository) Save(ctx context.Context, workflow *domain.Workflow
if workflow.Id == "" {
record = core.NewRecord(collection)
} else {
record, err = app.GetApp().FindRecordById(domain.CollectionNameWorkflow, workflow.Id)
record, err = app.GetApp().FindRecordById(collection, workflow.Id)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return workflow, domain.ErrRecordNotFound
@@ -85,7 +85,6 @@ func (r *WorkflowRepository) Save(ctx context.Context, workflow *domain.Workflow
record.Set("lastRunId", workflow.LastRunId)
record.Set("lastRunStatus", string(workflow.LastRunStatus))
record.Set("lastRunTime", workflow.LastRunTime)
if err := app.GetApp().Save(record); err != nil {
return workflow, err
}
@@ -96,65 +95,6 @@ func (r *WorkflowRepository) Save(ctx context.Context, workflow *domain.Workflow
return workflow, nil
}
func (r *WorkflowRepository) SaveRun(ctx context.Context, workflowRun *domain.WorkflowRun) (*domain.WorkflowRun, error) {
collection, err := app.GetApp().FindCollectionByNameOrId(domain.CollectionNameWorkflowRun)
if err != nil {
return workflowRun, err
}
var workflowRunRecord *core.Record
if workflowRun.Id == "" {
workflowRunRecord = core.NewRecord(collection)
} else {
workflowRunRecord, err = app.GetApp().FindRecordById(domain.CollectionNameWorkflowRun, workflowRun.Id)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return workflowRun, err
}
workflowRunRecord = core.NewRecord(collection)
}
}
err = app.GetApp().RunInTransaction(func(txApp core.App) error {
workflowRunRecord.Set("workflowId", workflowRun.WorkflowId)
workflowRunRecord.Set("trigger", string(workflowRun.Trigger))
workflowRunRecord.Set("status", string(workflowRun.Status))
workflowRunRecord.Set("startedAt", workflowRun.StartedAt)
workflowRunRecord.Set("endedAt", workflowRun.EndedAt)
workflowRunRecord.Set("logs", workflowRun.Logs)
workflowRunRecord.Set("error", workflowRun.Error)
err = txApp.Save(workflowRunRecord)
if err != nil {
return err
}
workflowRecord, err := txApp.FindRecordById(domain.CollectionNameWorkflow, workflowRun.WorkflowId)
if err != nil {
return err
}
workflowRecord.IgnoreUnchangedFields(true)
workflowRecord.Set("lastRunId", workflowRunRecord.Id)
workflowRecord.Set("lastRunStatus", workflowRunRecord.GetString("status"))
workflowRecord.Set("lastRunTime", workflowRunRecord.GetString("startedAt"))
err = txApp.Save(workflowRecord)
if err != nil {
return err
}
workflowRun.Id = workflowRunRecord.Id
workflowRun.CreatedAt = workflowRunRecord.GetDateTime("created").Time()
workflowRun.UpdatedAt = workflowRunRecord.GetDateTime("updated").Time()
return nil
})
if err != nil {
return workflowRun, err
}
return workflowRun, nil
}
func (r *WorkflowRepository) castRecordToModel(record *core.Record) (*domain.Workflow, error) {
if record == nil {
return nil, fmt.Errorf("record is nil")

166
internal/repository/workflow_output.go
View File

@@ -4,6 +4,7 @@ import (
"context"
"database/sql"
"errors"
"fmt"
"github.com/pocketbase/dbx"
"github.com/pocketbase/pocketbase/core"
@@ -17,13 +18,13 @@ func NewWorkflowOutputRepository() *WorkflowOutputRepository {
return &WorkflowOutputRepository{}
}
func (r *WorkflowOutputRepository) GetByNodeId(ctx context.Context, nodeId string) (*domain.WorkflowOutput, error) {
func (r *WorkflowOutputRepository) GetByNodeId(ctx context.Context, workflowNodeId string) (*domain.WorkflowOutput, error) {
records, err := app.GetApp().FindRecordsByFilter(
domain.CollectionNameWorkflowOutput,
"nodeId={:nodeId}",
"-created",
1, 0,
dbx.Params{"nodeId": nodeId},
dbx.Params{"nodeId": workflowNodeId},
)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
@@ -34,103 +35,128 @@ func (r *WorkflowOutputRepository) GetByNodeId(ctx context.Context, nodeId strin
if len(records) == 0 {
return nil, domain.ErrRecordNotFound
}
record := records[0]
return r.castRecordToModel(records[0])
}
func (r *WorkflowOutputRepository) Save(ctx context.Context, workflowOutput *domain.WorkflowOutput) (*domain.WorkflowOutput, error) {
record, err := r.saveRecord(workflowOutput)
if err != nil {
return workflowOutput, err
}
workflowOutput.Id = record.Id
workflowOutput.CreatedAt = record.GetDateTime("created").Time()
workflowOutput.UpdatedAt = record.GetDateTime("updated").Time()
return workflowOutput, nil
}
func (r *WorkflowOutputRepository) SaveWithCertificate(ctx context.Context, workflowOutput *domain.WorkflowOutput, certificate *domain.Certificate) (*domain.WorkflowOutput, error) {
record, err := r.saveRecord(workflowOutput)
if err != nil {
return workflowOutput, err
} else {
workflowOutput.Id = record.Id
workflowOutput.CreatedAt = record.GetDateTime("created").Time()
workflowOutput.UpdatedAt = record.GetDateTime("updated").Time()
}
if certificate == nil {
panic("certificate is nil")
} else {
if certificate.WorkflowId != "" && certificate.WorkflowId != workflowOutput.WorkflowId {
return workflowOutput, fmt.Errorf("certificate #%s is not belong to workflow #%s", certificate.Id, workflowOutput.WorkflowId)
}
if certificate.WorkflowRunId != "" && certificate.WorkflowRunId != workflowOutput.RunId {
return workflowOutput, fmt.Errorf("certificate #%s is not belong to workflow run #%s", certificate.Id, workflowOutput.RunId)
}
if certificate.WorkflowNodeId != "" && certificate.WorkflowNodeId != workflowOutput.NodeId {
return workflowOutput, fmt.Errorf("certificate #%s is not belong to workflow node #%s", certificate.Id, workflowOutput.NodeId)
}
if certificate.WorkflowOutputId != "" && certificate.WorkflowOutputId != workflowOutput.Id {
return workflowOutput, fmt.Errorf("certificate #%s is not belong to workflow output #%s", certificate.Id, workflowOutput.Id)
}
certificate.WorkflowId = workflowOutput.WorkflowId
certificate.WorkflowRunId = workflowOutput.RunId
certificate.WorkflowNodeId = workflowOutput.NodeId
certificate.WorkflowOutputId = workflowOutput.Id
certificate, err := NewCertificateRepository().Save(ctx, certificate)
if err != nil {
return workflowOutput, err
}
// 写入证书 ID 到工作流输出结果中
for i, item := range workflowOutput.Outputs {
if item.Name == string(domain.WorkflowNodeIONameCertificate) {
workflowOutput.Outputs[i].Value = certificate.Id
break
}
}
record.Set("outputs", workflowOutput.Outputs)
if err := app.GetApp().Save(record); err != nil {
return workflowOutput, err
}
}
return workflowOutput, err
}
func (r *WorkflowOutputRepository) castRecordToModel(record *core.Record) (*domain.WorkflowOutput, error) {
if record == nil {
return nil, fmt.Errorf("record is nil")
}
node := &domain.WorkflowNode{}
if err := record.UnmarshalJSONField("node", node); err != nil {
return nil, errors.New("failed to unmarshal node")
return nil, err
}
outputs := make([]domain.WorkflowNodeIO, 0)
if err := record.UnmarshalJSONField("outputs", &outputs); err != nil {
return nil, errors.New("failed to unmarshal output")
return nil, err
}
rs := &domain.WorkflowOutput{
workflowOutput := &domain.WorkflowOutput{
Meta: domain.Meta{
Id: record.Id,
CreatedAt: record.GetDateTime("created").Time(),
UpdatedAt: record.GetDateTime("updated").Time(),
},
WorkflowId: record.GetString("workflowId"),
RunId: record.GetString("runId"),
NodeId: record.GetString("nodeId"),
Node: node,
Outputs: outputs,
Succeeded: record.GetBool("succeeded"),
}
return rs, nil
return workflowOutput, nil
}
// 保存节点输出
func (r *WorkflowOutputRepository) Save(ctx context.Context, output *domain.WorkflowOutput, certificate *domain.Certificate, cb func(id string) error) error {
var record *core.Record
var err error
func (r *WorkflowOutputRepository) saveRecord(workflowOutput *domain.WorkflowOutput) (*core.Record, error) {
collection, err := app.GetApp().FindCollectionByNameOrId(domain.CollectionNameWorkflowOutput)
if err != nil {
return nil, err
}
if output.Id == "" {
collection, err := app.GetApp().FindCollectionByNameOrId(domain.CollectionNameWorkflowOutput)
if err != nil {
return err
}
var record *core.Record
if workflowOutput.Id == "" {
record = core.NewRecord(collection)
} else {
record, err = app.GetApp().FindRecordById(domain.CollectionNameWorkflowOutput, output.Id)
record, err = app.GetApp().FindRecordById(collection, workflowOutput.Id)
if err != nil {
return err
return record, err
}
}
record.Set("workflowId", output.WorkflowId)
record.Set("nodeId", output.NodeId)
record.Set("node", output.Node)
record.Set("outputs", output.Outputs)
record.Set("succeeded", output.Succeeded)
record.Set("workflowId", workflowOutput.WorkflowId)
record.Set("runId", workflowOutput.RunId)
record.Set("nodeId", workflowOutput.NodeId)
record.Set("node", workflowOutput.Node)
record.Set("outputs", workflowOutput.Outputs)
record.Set("succeeded", workflowOutput.Succeeded)
if err := app.GetApp().Save(record); err != nil {
return err
return record, err
}
if cb != nil && certificate != nil {
if err := cb(record.Id); err != nil {
return err
}
certCollection, err := app.GetApp().FindCollectionByNameOrId(domain.CollectionNameCertificate)
if err != nil {
return err
}
certRecord := core.NewRecord(certCollection)
certRecord.Set("source", string(certificate.Source))
certRecord.Set("subjectAltNames", certificate.SubjectAltNames)
certRecord.Set("certificate", certificate.Certificate)
certRecord.Set("privateKey", certificate.PrivateKey)
certRecord.Set("issuerCertificate", certificate.IssuerCertificate)
certRecord.Set("effectAt", certificate.EffectAt)
certRecord.Set("expireAt", certificate.ExpireAt)
certRecord.Set("acmeCertUrl", certificate.ACMECertUrl)
certRecord.Set("acmeCertStableUrl", certificate.ACMECertStableUrl)
certRecord.Set("workflowId", certificate.WorkflowId)
certRecord.Set("workflowNodeId", certificate.WorkflowNodeId)
certRecord.Set("workflowOutputId", certificate.WorkflowOutputId)
if err := app.GetApp().Save(certRecord); err != nil {
return err
}
// 更新 certificate
for i, item := range output.Outputs {
if item.Name == string(domain.WorkflowNodeIONameCertificate) {
output.Outputs[i].Value = certRecord.Id
break
}
}
record.Set("outputs", output.Outputs)
if err := app.GetApp().Save(record); err != nil {
return err
}
}
return nil
return record, nil
}

124
internal/repository/workflow_run.go Normal file
View File

@@ -0,0 +1,124 @@
package repository
import (
"context"
"database/sql"
"errors"
"fmt"
"github.com/pocketbase/pocketbase/core"
"github.com/usual2970/certimate/internal/app"
"github.com/usual2970/certimate/internal/domain"
)
type WorkflowRunRepository struct{}
func NewWorkflowRunRepository() *WorkflowRunRepository {
return &WorkflowRunRepository{}
}
func (r *WorkflowRunRepository) GetById(ctx context.Context, id string) (*domain.WorkflowRun, error) {
record, err := app.GetApp().FindRecordById(domain.CollectionNameWorkflowRun, id)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return nil, domain.ErrRecordNotFound
}
return nil, err
}
return r.castRecordToModel(record)
}
func (r *WorkflowRunRepository) Save(ctx context.Context, workflowRun *domain.WorkflowRun) (*domain.WorkflowRun, error) {
collection, err := app.GetApp().FindCollectionByNameOrId(domain.CollectionNameWorkflowRun)
if err != nil {
return workflowRun, err
}
var record *core.Record
if workflowRun.Id == "" {
record = core.NewRecord(collection)
} else {
record, err = app.GetApp().FindRecordById(collection, workflowRun.Id)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return workflowRun, err
}
record = core.NewRecord(collection)
}
}
err = app.GetApp().RunInTransaction(func(txApp core.App) error {
record.Set("workflowId", workflowRun.WorkflowId)
record.Set("trigger", string(workflowRun.Trigger))
record.Set("status", string(workflowRun.Status))
record.Set("startedAt", workflowRun.StartedAt)
record.Set("endedAt", workflowRun.EndedAt)
record.Set("logs", workflowRun.Logs)
record.Set("error", workflowRun.Error)
err = txApp.Save(record)
if err != nil {
return err
}
workflowRun.Id = record.Id
workflowRun.CreatedAt = record.GetDateTime("created").Time()
workflowRun.UpdatedAt = record.GetDateTime("updated").Time()
// 事务级联更新所属工作流的最后运行记录
workflowRecord, err := txApp.FindRecordById(domain.CollectionNameWorkflow, workflowRun.WorkflowId)
if err != nil {
return err
} else if workflowRun.Id == workflowRecord.GetString("lastRunId") {
workflowRecord.IgnoreUnchangedFields(true)
workflowRecord.Set("lastRunStatus", record.GetString("status"))
err = txApp.Save(workflowRecord)
if err != nil {
return err
}
} else if workflowRecord.GetDateTime("lastRunTime").Time().IsZero() || workflowRun.StartedAt.After(workflowRecord.GetDateTime("lastRunTime").Time()) {
workflowRecord.IgnoreUnchangedFields(true)
workflowRecord.Set("lastRunId", record.Id)
workflowRecord.Set("lastRunStatus", record.GetString("status"))
workflowRecord.Set("lastRunTime", record.GetString("startedAt"))
err = txApp.Save(workflowRecord)
if err != nil {
return err
}
}
return nil
})
if err != nil {
return workflowRun, err
}
return workflowRun, nil
}
func (r *WorkflowRunRepository) castRecordToModel(record *core.Record) (*domain.WorkflowRun, error) {
if record == nil {
return nil, fmt.Errorf("record is nil")
}
logs := make([]domain.WorkflowRunLog, 0)
if err := record.UnmarshalJSONField("logs", &logs); err != nil {
return nil, err
}
workflowRun := &domain.WorkflowRun{
Meta: domain.Meta{
Id: record.Id,
CreatedAt: record.GetDateTime("created").Time(),
UpdatedAt: record.GetDateTime("updated").Time(),
},
WorkflowId: record.GetString("workflowId"),
Status: domain.WorkflowRunStatusType(record.GetString("status")),
Trigger: domain.WorkflowTriggerType(record.GetString("trigger")),
StartedAt: record.GetDateTime("startedAt").Time(),
EndedAt: record.GetDateTime("endedAt").Time(),
Logs: logs,
Error: record.GetString("error"),
}
return workflowRun, nil
}

16
internal/rest/handlers/certificate.go
View File

@@ -11,9 +11,9 @@ import (
)
type certificateService interface {
ArchiveFile(ctx context.Context, req *dtos.CertificateArchiveFileReq) ([]byte, error)
ArchiveFile(ctx context.Context, req *dtos.CertificateArchiveFileReq) (*dtos.CertificateArchiveFileResp, error)
ValidateCertificate(ctx context.Context, req *dtos.CertificateValidateCertificateReq) (*dtos.CertificateValidateCertificateResp, error)
ValidatePrivateKey(ctx context.Context, req *dtos.CertificateValidatePrivateKeyReq) error
ValidatePrivateKey(ctx context.Context, req *dtos.CertificateValidatePrivateKeyReq) (*dtos.CertificateValidatePrivateKeyResp, error)
}
type CertificateHandler struct {
@@ -38,10 +38,10 @@ func (handler *CertificateHandler) archiveFile(e *core.RequestEvent) error {
return resp.Err(e, err)
}
if bt, err := handler.service.ArchiveFile(e.Request.Context(), req); err != nil {
if res, err := handler.service.ArchiveFile(e.Request.Context(), req); err != nil {
return resp.Err(e, err)
} else {
return resp.Ok(e, bt)
return resp.Ok(e, res)
}
}
@@ -51,10 +51,10 @@ func (handler *CertificateHandler) validateCertificate(e *core.RequestEvent) err
return resp.Err(e, err)
}
if rs, err := handler.service.ValidateCertificate(e.Request.Context(), req); err != nil {
if res, err := handler.service.ValidateCertificate(e.Request.Context(), req); err != nil {
return resp.Err(e, err)
} else {
return resp.Ok(e, rs)
return resp.Ok(e, res)
}
}
@@ -64,9 +64,9 @@ func (handler *CertificateHandler) validatePrivateKey(e *core.RequestEvent) erro
return resp.Err(e, err)
}
if err := handler.service.ValidatePrivateKey(e.Request.Context(), req); err != nil {
if res, err := handler.service.ValidatePrivateKey(e.Request.Context(), req); err != nil {
return resp.Err(e, err)
} else {
return resp.Ok(e, nil)
return resp.Ok(e, res)
}
}

2
internal/rest/handlers/workflow.go
View File

@@ -13,7 +13,7 @@ import (
type workflowService interface {
StartRun(ctx context.Context, req *dtos.WorkflowStartRunReq) error
CancelRun(ctx context.Context, req *dtos.WorkflowCancelRunReq) error
Stop(ctx context.Context)
Shutdown(ctx context.Context)
}
type WorkflowHandler struct {

9
internal/rest/routes/routes.go
View File

@@ -27,13 +27,14 @@ func Register(router *router.Router[*core.RequestEvent]) {
certificateSvc = certificate.NewCertificateService(certificateRepo)
workflowRepo := repository.NewWorkflowRepository()
workflowSvc = workflow.NewWorkflowService(workflowRepo)
workflowRunRepo := repository.NewWorkflowRunRepository()
workflowSvc = workflow.NewWorkflowService(workflowRepo, workflowRunRepo)
statisticsRepo := repository.NewStatisticsRepository()
statisticsSvc = statistics.NewStatisticsService(statisticsRepo)
notifyRepo := repository.NewSettingsRepository()
notifySvc = notify.NewNotifyService(notifyRepo)
settingsRepo := repository.NewSettingsRepository()
notifySvc = notify.NewNotifyService(settingsRepo)
group := router.Group("/api")
group.Bind(apis.RequireSuperuserAuth())
@@ -45,6 +46,6 @@ func Register(router *router.Router[*core.RequestEvent]) {
func Unregister() {
if workflowSvc != nil {
workflowSvc.Stop(context.Background())
workflowSvc.Shutdown(context.Background())
}
}

2
internal/scheduler/certificate.go
View File

@@ -6,6 +6,6 @@ type certificateService interface {
InitSchedule(ctx context.Context) error
}
func NewCertificateScheduler(service certificateService) error {
func InitCertificateScheduler(service certificateService) error {
return service.InitSchedule(context.Background())
}

12
internal/scheduler/scheduler.go
View File

@@ -1,6 +1,7 @@
package scheduler
import (
"github.com/usual2970/certimate/internal/app"
"github.com/usual2970/certimate/internal/certificate"
"github.com/usual2970/certimate/internal/repository"
"github.com/usual2970/certimate/internal/workflow"
@@ -8,12 +9,17 @@ import (
func Register() {
workflowRepo := repository.NewWorkflowRepository()
workflowSvc := workflow.NewWorkflowService(workflowRepo)
workflowRunRepo := repository.NewWorkflowRunRepository()
workflowSvc := workflow.NewWorkflowService(workflowRepo, workflowRunRepo)
certificateRepo := repository.NewCertificateRepository()
certificateSvc := certificate.NewCertificateService(certificateRepo)
NewCertificateScheduler(certificateSvc)
if err := InitWorkflowScheduler(workflowSvc); err != nil {
app.GetLogger().Error("failed to init workflow scheduler", "err", err)
}
NewWorkflowScheduler(workflowSvc)
if err := InitCertificateScheduler(certificateSvc); err != nil {
app.GetLogger().Error("failed to init certificate scheduler", "err", err)
}
}

2
internal/scheduler/workflow.go
View File

@@ -6,6 +6,6 @@ type workflowService interface {
InitSchedule(ctx context.Context) error
}
func NewWorkflowScheduler(service workflowService) error {
func InitWorkflowScheduler(service workflowService) error {
return service.InitSchedule(context.Background())
}

8
internal/statistics/service.go
View File

@@ -11,15 +11,15 @@ type statisticsRepository interface {
}
type StatisticsService struct {
repo statisticsRepository
statRepo statisticsRepository
}
func NewStatisticsService(repo statisticsRepository) *StatisticsService {
func NewStatisticsService(statRepo statisticsRepository) *StatisticsService {
return &StatisticsService{
repo: repo,
statRepo: statRepo,
}
}
func (s *StatisticsService) Get(ctx context.Context) (*domain.Statistics, error) {
return s.repo.Get(ctx)
return s.statRepo.Get(ctx)
}

280
internal/workflow/dispatcher/dispatcher.go Normal file
View File

@@ -0,0 +1,280 @@
package dispatcher
import (
"context"
"errors"
"fmt"
"os"
"strconv"
"sync"
"time"
"github.com/usual2970/certimate/internal/app"
"github.com/usual2970/certimate/internal/domain"
"github.com/usual2970/certimate/internal/pkg/utils/slices"
)
var maxWorkers = 16
func init() {
envMaxWorkers := os.Getenv("CERTIMATE_WORKFLOW_MAX_WORKERS")
if n, err := strconv.Atoi(envMaxWorkers); err != nil && n > 0 {
maxWorkers = n
}
}
type workflowWorker struct {
Data *WorkflowWorkerData
Cancel context.CancelFunc
}
type WorkflowWorkerData struct {
WorkflowId string
WorkflowContent *domain.WorkflowNode
RunId string
}
type WorkflowDispatcher struct {
semaphore chan struct{}
queue []*WorkflowWorkerData
queueMutex sync.Mutex
workers map[string]*workflowWorker // key: WorkflowId
workerIdMap map[string]string // key: RunId, value: WorkflowId
workerMutex sync.Mutex
chWork chan *WorkflowWorkerData
chCandi chan struct{}
wg sync.WaitGroup
workflowRepo workflowRepository
workflowRunRepo workflowRunRepository
}
func newWorkflowDispatcher(workflowRepo workflowRepository, workflowRunRepo workflowRunRepository) *WorkflowDispatcher {
dispatcher := &WorkflowDispatcher{
semaphore: make(chan struct{}, maxWorkers),
queue: make([]*WorkflowWorkerData, 0),
queueMutex: sync.Mutex{},
workers: make(map[string]*workflowWorker),
workerIdMap: make(map[string]string),
workerMutex: sync.Mutex{},
chWork: make(chan *WorkflowWorkerData),
chCandi: make(chan struct{}, 1),
workflowRepo: workflowRepo,
workflowRunRepo: workflowRunRepo,
}
go func() {
for {
select {
case <-dispatcher.chWork:
dispatcher.dequeueWorker()
case <-dispatcher.chCandi:
dispatcher.dequeueWorker()
}
}
}()
return dispatcher
}
func (w *WorkflowDispatcher) Dispatch(data *WorkflowWorkerData) {
if data == nil {
panic("worker data is nil")
}
w.enqueueWorker(data)
select {
case w.chWork <- data:
default:
}
}
func (w *WorkflowDispatcher) Cancel(runId string) {
hasWorker := false
// 取消正在执行的 WorkflowRun
w.workerMutex.Lock()
if workflowId, ok := w.workerIdMap[runId]; ok {
if worker, ok := w.workers[workflowId]; ok {
hasWorker = true
worker.Cancel()
delete(w.workers, workflowId)
delete(w.workerIdMap, runId)
}
}
w.workerMutex.Unlock()
// 移除排队中的 WorkflowRun
w.queueMutex.Lock()
w.queue = slices.Filter(w.queue, func(d *WorkflowWorkerData) bool {
return d.RunId != runId
})
w.queueMutex.Unlock()
// 已挂起，查询 WorkflowRun 并更新其状态为 Canceled
if !hasWorker {
if run, err := w.workflowRunRepo.GetById(context.Background(), runId); err == nil {
if run.Status == domain.WorkflowRunStatusTypePending || run.Status == domain.WorkflowRunStatusTypeRunning {
run.Status = domain.WorkflowRunStatusTypeCanceled
w.workflowRunRepo.Save(context.Background(), run)
}
}
}
}
func (w *WorkflowDispatcher) Shutdown() {
// 清空排队中的 WorkflowRun
w.queueMutex.Lock()
w.queue = make([]*WorkflowWorkerData, 0)
w.queueMutex.Unlock()
// 等待所有正在执行的 WorkflowRun 完成
w.workerMutex.Lock()
for _, worker := range w.workers {
worker.Cancel()
delete(w.workers, worker.Data.WorkflowId)
delete(w.workerIdMap, worker.Data.RunId)
}
w.workerMutex.Unlock()
w.wg.Wait()
}
func (w *WorkflowDispatcher) enqueueWorker(data *WorkflowWorkerData) {
w.queueMutex.Lock()
defer w.queueMutex.Unlock()
w.queue = append(w.queue, data)
}
func (w *WorkflowDispatcher) dequeueWorker() {
for {
select {
case w.semaphore <- struct{}{}:
default:
// 达到最大并发数
return
}
w.queueMutex.Lock()
if len(w.queue) == 0 {
w.queueMutex.Unlock()
<-w.semaphore
return
}
data := w.queue[0]
w.queue = w.queue[1:]
w.queueMutex.Unlock()
// 检查是否有相同 WorkflowId 的 WorkflowRun 正在执行
// 如果有，则重新排队，以保证同一个工作流同一时间内只有一个正在执行
// 即不同 WorkflowId 的任务并行化，相同 WorkflowId 的任务串行化
w.workerMutex.Lock()
if _, exists := w.workers[data.WorkflowId]; exists {
w.queueMutex.Lock()
w.queue = append(w.queue, data)
w.queueMutex.Unlock()
w.workerMutex.Unlock()
<-w.semaphore
continue
}
ctx, cancel := context.WithCancel(context.Background())
w.workers[data.WorkflowId] = &workflowWorker{data, cancel}
w.workerIdMap[data.RunId] = data.WorkflowId
w.workerMutex.Unlock()
w.wg.Add(1)
go w.work(ctx, data)
}
}
func (w *WorkflowDispatcher) work(ctx context.Context, data *WorkflowWorkerData) {
defer func() {
<-w.semaphore
w.workerMutex.Lock()
delete(w.workers, data.WorkflowId)
delete(w.workerIdMap, data.RunId)
w.workerMutex.Unlock()
w.wg.Done()
// 尝试取出排队中的其他 WorkflowRun 继续执行
select {
case w.chCandi <- struct{}{}:
default:
}
}()
// 查询 WorkflowRun
run, err := w.workflowRunRepo.GetById(ctx, data.RunId)
if err != nil {
if !(errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) {
app.GetLogger().Error(fmt.Sprintf("failed to get workflow run #%s", data.RunId), "err", err)
}
return
} else if run.Status != domain.WorkflowRunStatusTypePending {
return
} else if ctx.Err() != nil {
run.Status = domain.WorkflowRunStatusTypeCanceled
w.workflowRunRepo.Save(ctx, run)
return
}
// 更新 WorkflowRun 状态为 Running
run.Status = domain.WorkflowRunStatusTypeRunning
if _, err := w.workflowRunRepo.Save(ctx, run); err != nil {
if !(errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) {
panic(err)
}
return
}
// 执行工作流
invoker := newWorkflowInvokerWithData(w.workflowRunRepo, data)
if runErr := invoker.Invoke(ctx); runErr != nil {
if errors.Is(runErr, context.Canceled) {
run.Status = domain.WorkflowRunStatusTypeCanceled
run.Logs = invoker.GetLogs()
} else {
run.Status = domain.WorkflowRunStatusTypeFailed
run.EndedAt = time.Now()
run.Logs = invoker.GetLogs()
run.Error = runErr.Error()
}
if _, err := w.workflowRunRepo.Save(ctx, run); err != nil {
if !(errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) {
panic(err)
}
}
return
}
// 更新 WorkflowRun 状态为 Succeeded/Failed
run.EndedAt = time.Now()
run.Logs = invoker.GetLogs()
run.Error = domain.WorkflowRunLogs(invoker.GetLogs()).ErrorString()
if run.Error == "" {
run.Status = domain.WorkflowRunStatusTypeSucceeded
} else {
run.Status = domain.WorkflowRunStatusTypeFailed
}
if _, err := w.workflowRunRepo.Save(ctx, run); err != nil {
if !(errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) {
panic(err)
}
}
}

115
internal/workflow/dispatcher/invoker.go Normal file
View File

@@ -0,0 +1,115 @@
package dispatcher
import (
"context"
"errors"
"github.com/usual2970/certimate/internal/domain"
nodes "github.com/usual2970/certimate/internal/workflow/node-processor"
)
type workflowInvoker struct {
workflowId string
workflowContent *domain.WorkflowNode
runId string
runLogs []domain.WorkflowRunLog
workflowRunRepo workflowRunRepository
}
func newWorkflowInvokerWithData(workflowRunRepo workflowRunRepository, data *WorkflowWorkerData) *workflowInvoker {
if data == nil {
panic("worker data is nil")
}
// TODO: 待优化，日志与执行解耦
return &workflowInvoker{
workflowId: data.WorkflowId,
workflowContent: data.WorkflowContent,
runId: data.RunId,
runLogs: make([]domain.WorkflowRunLog, 0),
workflowRunRepo: workflowRunRepo,
}
}
func (w *workflowInvoker) Invoke(ctx context.Context) error {
ctx = context.WithValue(ctx, "workflow_id", w.workflowId)
ctx = context.WithValue(ctx, "workflow_run_id", w.runId)
return w.processNode(ctx, w.workflowContent)
}
func (w *workflowInvoker) GetLogs() []domain.WorkflowRunLog {
return w.runLogs
}
func (w *workflowInvoker) processNode(ctx context.Context, node *domain.WorkflowNode) error {
current := node
for current != nil {
if ctx.Err() != nil {
return ctx.Err()
}
if current.Type == domain.WorkflowNodeTypeBranch || current.Type == domain.WorkflowNodeTypeExecuteResultBranch {
for _, branch := range current.Branches {
if err := w.processNode(ctx, &branch); err != nil {
// 并行分支的某一分支发生错误时，忽略此错误，继续执行其他分支
if !(errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) {
continue
}
return err
}
}
}
var processor nodes.NodeProcessor
var procErr error
for {
if current.Type != domain.WorkflowNodeTypeBranch && current.Type != domain.WorkflowNodeTypeExecuteResultBranch {
processor, procErr = nodes.GetProcessor(current)
if procErr != nil {
break
}
procErr = processor.Process(ctx)
log := processor.GetLog(ctx)
if log != nil {
w.runLogs = append(w.runLogs, *log)
// TODO: 待优化，把 /pkg/core/* 包下的输出写入到 DEBUG 级别的日志中
if run, err := w.workflowRunRepo.GetById(ctx, w.runId); err == nil {
run.Logs = w.runLogs
w.workflowRunRepo.Save(ctx, run)
}
}
if procErr != nil {
break
}
}
break
}
// TODO: 优化可读性
if procErr != nil && current.Next != nil && current.Next.Type != domain.WorkflowNodeTypeExecuteResultBranch {
return procErr
} else if procErr != nil && current.Next != nil && current.Next.Type == domain.WorkflowNodeTypeExecuteResultBranch {
current = w.getBranchByType(current.Next.Branches, domain.WorkflowNodeTypeExecuteFailure)
} else if procErr == nil && current.Next != nil && current.Next.Type == domain.WorkflowNodeTypeExecuteResultBranch {
current = w.getBranchByType(current.Next.Branches, domain.WorkflowNodeTypeExecuteSuccess)
} else {
current = current.Next
}
}
return nil
}
func (w *workflowInvoker) getBranchByType(branches []domain.WorkflowNode, nodeType domain.WorkflowNodeType) *domain.WorkflowNode {
for _, branch := range branches {
if branch.Type == nodeType {
return &branch
}
}
return nil
}

32
internal/workflow/dispatcher/singleton.go Normal file
View File

@@ -0,0 +1,32 @@
package dispatcher
import (
"context"
"sync"
"github.com/usual2970/certimate/internal/domain"
)
type workflowRepository interface {
GetById(ctx context.Context, id string) (*domain.Workflow, error)
Save(ctx context.Context, workflow *domain.Workflow) (*domain.Workflow, error)
}
type workflowRunRepository interface {
GetById(ctx context.Context, id string) (*domain.WorkflowRun, error)
Save(ctx context.Context, workflowRun *domain.WorkflowRun) (*domain.WorkflowRun, error)
}
var (
instance *WorkflowDispatcher
intanceOnce sync.Once
)
func GetSingletonDispatcher(workflowRepo workflowRepository, workflowRunRepo workflowRunRepository) *WorkflowDispatcher {
// TODO: 待优化构造过程
intanceOnce.Do(func() {
instance = newWorkflowDispatcher(workflowRepo, workflowRunRepo)
})
return instance
}

6
internal/workflow/event.go
View File

@@ -65,13 +65,13 @@ func onWorkflowRecordCreateOrUpdate(ctx context.Context, record *core.Record) er
// 反之，重新添加定时任务
err := scheduler.Add(fmt.Sprintf("workflow#%s", workflowId), record.GetString("triggerCron"), func() {
NewWorkflowService(repository.NewWorkflowRepository()).StartRun(ctx, &dtos.WorkflowStartRunReq{
workflowSrv := NewWorkflowService(repository.NewWorkflowRepository(), repository.NewWorkflowRunRepository())
workflowSrv.StartRun(ctx, &dtos.WorkflowStartRunReq{
WorkflowId: workflowId,
Trigger: domain.WorkflowTriggerTypeAuto,
RunTrigger: domain.WorkflowTriggerTypeAuto,
})
})
if err != nil {
app.GetLogger().Error("add cron job failed", "err", err)
return fmt.Errorf("add cron job failed: %w", err)
}

82
internal/workflow/node-processor/apply_node.go
View File

@@ -3,7 +3,6 @@ package nodeprocessor
import (
"context"
"fmt"
"strings"
"time"
"golang.org/x/exp/maps"
@@ -15,104 +14,93 @@ import (
)
type applyNode struct {
node *domain.WorkflowNode
node *domain.WorkflowNode
*nodeLogger
certRepo certificateRepository
outputRepo workflowOutputRepository
*nodeLogger
}
func NewApplyNode(node *domain.WorkflowNode) *applyNode {
return &applyNode{
node: node,
nodeLogger: NewNodeLogger(node),
outputRepo: repository.NewWorkflowOutputRepository(),
nodeLogger: newNodeLogger(node),
certRepo: repository.NewCertificateRepository(),
outputRepo: repository.NewWorkflowOutputRepository(),
}
}
// 申请节点根据申请类型执行不同的操作
func (a *applyNode) Run(ctx context.Context) error {
a.AddOutput(ctx, a.node.Name, "开始执行")
func (n *applyNode) Process(ctx context.Context) error {
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "进入申请证书节点")
// 查询上次执行结果
lastOutput, err := a.outputRepo.GetByNodeId(ctx, a.node.Id)
lastOutput, err := n.outputRepo.GetByNodeId(ctx, n.node.Id)
if err != nil && !domain.IsRecordNotFoundError(err) {
a.AddOutput(ctx, a.node.Name, "查询申请记录失败", err.Error())
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "查询申请记录失败", err.Error())
return err
}
// 检测是否可以跳过本次执行
if skippable, skipReason := a.checkCanSkip(ctx, lastOutput); skippable {
a.AddOutput(ctx, a.node.Name, skipReason)
if skippable, skipReason := n.checkCanSkip(ctx, lastOutput); skippable {
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, skipReason)
return nil
}
// 初始化申请器
applicant, err := applicant.NewWithApplyNode(a.node)
applicant, err := applicant.NewWithApplyNode(n.node)
if err != nil {
a.AddOutput(ctx, a.node.Name, "获取申请对象失败", err.Error())
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "获取申请对象失败", err.Error())
return err
}
// 申请证书
applyResult, err := applicant.Apply()
if err != nil {
a.AddOutput(ctx, a.node.Name, "申请失败", err.Error())
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "申请失败", err.Error())
return err
}
a.AddOutput(ctx, a.node.Name, "申请成功")
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "申请成功")
// 解析证书并生成实体
certX509, err := certs.ParseCertificateFromPEM(applyResult.CertificateFullChain)
if err != nil {
a.AddOutput(ctx, a.node.Name, "解析证书失败", err.Error())
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "解析证书失败", err.Error())
return err
}
certificate := &domain.Certificate{
Source: domain.CertificateSourceTypeWorkflow,
SubjectAltNames: strings.Join(certX509.DNSNames, ";"),
Certificate: applyResult.CertificateFullChain,
PrivateKey: applyResult.PrivateKey,
IssuerCertificate: applyResult.IssuerCertificate,
ACMEAccountUrl: applyResult.ACMEAccountUrl,
ACMECertUrl: applyResult.ACMECertUrl,
ACMECertStableUrl: applyResult.ACMECertStableUrl,
EffectAt: certX509.NotBefore,
ExpireAt: certX509.NotAfter,
WorkflowId: getContextWorkflowId(ctx),
WorkflowNodeId: a.node.Id,
}
certificate.PopulateFromX509(certX509)
// 保存执行结果
// TODO: 先保持一个节点始终只有一个输出，后续增加版本控制
currentOutput := &domain.WorkflowOutput{
output := &domain.WorkflowOutput{
WorkflowId: getContextWorkflowId(ctx),
NodeId: a.node.Id,
Node: a.node,
RunId: getContextWorkflowRunId(ctx),
NodeId: n.node.Id,
Node: n.node,
Succeeded: true,
Outputs: a.node.Outputs,
Outputs: n.node.Outputs,
}
if lastOutput != nil {
currentOutput.Id = lastOutput.Id
}
if err := a.outputRepo.Save(ctx, currentOutput, certificate, func(id string) error {
if certificate != nil {
certificate.WorkflowOutputId = id
}
return nil
}); err != nil {
a.AddOutput(ctx, a.node.Name, "保存申请记录失败", err.Error())
if _, err := n.outputRepo.SaveWithCertificate(ctx, output, certificate); err != nil {
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "保存申请记录失败", err.Error())
return err
}
a.AddOutput(ctx, a.node.Name, "保存申请记录成功")
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "保存申请记录成功")
return nil
}
func (a *applyNode) checkCanSkip(ctx context.Context, lastOutput *domain.WorkflowOutput) (skip bool, reason string) {
func (n *applyNode) checkCanSkip(ctx context.Context, lastOutput *domain.WorkflowOutput) (skip bool, reason string) {
if lastOutput != nil && lastOutput.Succeeded {
// 比较和上次申请时的关键配置（即影响证书签发的）参数是否一致
currentNodeConfig := a.node.GetConfigForApply()
currentNodeConfig := n.node.GetConfigForApply()
lastNodeConfig := lastOutput.Node.GetConfigForApply()
if currentNodeConfig.Domains != lastNodeConfig.Domains {
return false, "配置项变化：域名"
@@ -130,11 +118,13 @@ func (a *applyNode) checkCanSkip(ctx context.Context, lastOutput *domain.Workflo
return false, "配置项变化：数字签名算法"
}
lastCertificate, _ := a.certRepo.GetByWorkflowNodeId(ctx, a.node.Id)
renewalInterval := time.Duration(currentNodeConfig.SkipBeforeExpiryDays) * time.Hour * 24
expirationTime := time.Until(lastCertificate.ExpireAt)
if lastCertificate != nil && expirationTime > renewalInterval {
return true, fmt.Sprintf("已申请过证书，且证书尚未临近过期（到期尚余 %d 天，预计距 %d 天时续期）", int(expirationTime.Hours()/24), currentNodeConfig.SkipBeforeExpiryDays)
lastCertificate, _ := n.certRepo.GetByWorkflowNodeId(ctx, n.node.Id)
if lastCertificate != nil {
renewalInterval := time.Duration(currentNodeConfig.SkipBeforeExpiryDays) * time.Hour * 24
expirationTime := time.Until(lastCertificate.ExpireAt)
if expirationTime > renewalInterval {
return true, fmt.Sprintf("已申请过证书，且证书尚未临近过期（尚余 %d 天过期，不足 %d 天时续期）", int(expirationTime.Hours()/24), currentNodeConfig.SkipBeforeExpiryDays)
}
}
}

10
internal/workflow/node-processor/condition_node.go
View File

@@ -14,15 +14,11 @@ type conditionNode struct {
func NewConditionNode(node *domain.WorkflowNode) *conditionNode {
return &conditionNode{
node: node,
nodeLogger: NewNodeLogger(node),
nodeLogger: newNodeLogger(node),
}
}
// 条件节点没有任何操作
func (c *conditionNode) Run(ctx context.Context) error {
c.AddOutput(ctx,
c.node.Name,
"完成",
)
func (n *conditionNode) Process(ctx context.Context) error {
// 此类型节点不需要执行任何操作，直接返回
return nil
}

68
internal/workflow/node-processor/deploy_node.go
View File

@@ -12,96 +12,92 @@ import (
)
type deployNode struct {
node *domain.WorkflowNode
node *domain.WorkflowNode
*nodeLogger
certRepo certificateRepository
outputRepo workflowOutputRepository
*nodeLogger
}
func NewDeployNode(node *domain.WorkflowNode) *deployNode {
return &deployNode{
node: node,
nodeLogger: NewNodeLogger(node),
outputRepo: repository.NewWorkflowOutputRepository(),
nodeLogger: newNodeLogger(node),
certRepo: repository.NewCertificateRepository(),
outputRepo: repository.NewWorkflowOutputRepository(),
}
}
func (d *deployNode) Run(ctx context.Context) error {
d.AddOutput(ctx, d.node.Name, "开始执行")
func (n *deployNode) Process(ctx context.Context) error {
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "开始执行")
// 查询上次执行结果
lastOutput, err := d.outputRepo.GetByNodeId(ctx, d.node.Id)
lastOutput, err := n.outputRepo.GetByNodeId(ctx, n.node.Id)
if err != nil && !domain.IsRecordNotFoundError(err) {
d.AddOutput(ctx, d.node.Name, "查询部署记录失败", err.Error())
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "查询部署记录失败", err.Error())
return err
}
// 获取前序节点输出证书
previousNodeOutputCertificateSource := d.node.GetConfigForDeploy().Certificate
previousNodeOutputCertificateSource := n.node.GetConfigForDeploy().Certificate
previousNodeOutputCertificateSourceSlice := strings.Split(previousNodeOutputCertificateSource, "#")
if len(previousNodeOutputCertificateSourceSlice) != 2 {
d.AddOutput(ctx, d.node.Name, "证书来源配置错误", previousNodeOutputCertificateSource)
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "证书来源配置错误", previousNodeOutputCertificateSource)
return fmt.Errorf("证书来源配置错误: %s", previousNodeOutputCertificateSource)
}
certificate, err := d.certRepo.GetByWorkflowNodeId(ctx, previousNodeOutputCertificateSourceSlice[0])
certificate, err := n.certRepo.GetByWorkflowNodeId(ctx, previousNodeOutputCertificateSourceSlice[0])
if err != nil {
d.AddOutput(ctx, d.node.Name, "获取证书失败", err.Error())
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "获取证书失败", err.Error())
return err
}
// 检测是否可以跳过本次执行
if skippable, skipReason := d.checkCanSkip(ctx, lastOutput); skippable {
if certificate.CreatedAt.Before(lastOutput.UpdatedAt) {
d.AddOutput(ctx, d.node.Name, "已部署过且证书未更新")
} else {
d.AddOutput(ctx, d.node.Name, skipReason)
if lastOutput != nil && certificate.CreatedAt.Before(lastOutput.UpdatedAt) {
if skippable, skipReason := n.checkCanSkip(ctx, lastOutput); skippable {
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, skipReason)
return nil
}
return nil
}
// 初始化部署器
deploy, err := deployer.NewWithDeployNode(d.node, struct {
deployer, err := deployer.NewWithDeployNode(n.node, struct {
Certificate string
PrivateKey string
}{Certificate: certificate.Certificate, PrivateKey: certificate.PrivateKey})
if err != nil {
d.AddOutput(ctx, d.node.Name, "获取部署对象失败", err.Error())
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "获取部署对象失败", err.Error())
return err
}
// 部署证书
if err := deploy.Deploy(ctx); err != nil {
d.AddOutput(ctx, d.node.Name, "部署失败", err.Error())
if err := deployer.Deploy(ctx); err != nil {
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "部署失败", err.Error())
return err
}
d.AddOutput(ctx, d.node.Name, "部署成功")
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "部署成功")
// 保存执行结果
// TODO: 先保持一个节点始终只有一个输出，后续增加版本控制
currentOutput := &domain.WorkflowOutput{
Meta: domain.Meta{},
output := &domain.WorkflowOutput{
WorkflowId: getContextWorkflowId(ctx),
NodeId: d.node.Id,
Node: d.node,
RunId: getContextWorkflowRunId(ctx),
NodeId: n.node.Id,
Node: n.node,
Succeeded: true,
}
if lastOutput != nil {
currentOutput.Id = lastOutput.Id
}
if err := d.outputRepo.Save(ctx, currentOutput, nil, nil); err != nil {
d.AddOutput(ctx, d.node.Name, "保存部署记录失败", err.Error())
if _, err := n.outputRepo.Save(ctx, output); err != nil {
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "保存部署记录失败", err.Error())
return err
}
d.AddOutput(ctx, d.node.Name, "保存部署记录成功")
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "保存部署记录成功")
return nil
}
func (d *deployNode) checkCanSkip(ctx context.Context, lastOutput *domain.WorkflowOutput) (skip bool, reason string) {
func (n *deployNode) checkCanSkip(ctx context.Context, lastOutput *domain.WorkflowOutput) (skip bool, reason string) {
if lastOutput != nil && lastOutput.Succeeded {
// 比较和上次部署时的关键配置（即影响证书部署的）参数是否一致
currentNodeConfig := d.node.GetConfigForDeploy()
currentNodeConfig := n.node.GetConfigForDeploy()
lastNodeConfig := lastOutput.Node.GetConfigForDeploy()
if currentNodeConfig.ProviderAccessId != lastNodeConfig.ProviderAccessId {
return false, "配置项变化：主机提供商授权"

11
internal/workflow/node-processor/execute_failure_node.go
View File

@@ -14,14 +14,13 @@ type executeFailureNode struct {
func NewExecuteFailureNode(node *domain.WorkflowNode) *executeFailureNode {
return &executeFailureNode{
node: node,
nodeLogger: NewNodeLogger(node),
nodeLogger: newNodeLogger(node),
}
}
func (e *executeFailureNode) Run(ctx context.Context) error {
e.AddOutput(ctx,
e.node.Name,
"进入执行失败分支",
)
func (n *executeFailureNode) Process(ctx context.Context) error {
// 此类型节点不需要执行任何操作，直接返回
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "进入执行失败分支")
return nil
}

11
internal/workflow/node-processor/execute_success_node.go
View File

@@ -14,14 +14,13 @@ type executeSuccessNode struct {
func NewExecuteSuccessNode(node *domain.WorkflowNode) *executeSuccessNode {
return &executeSuccessNode{
node: node,
nodeLogger: NewNodeLogger(node),
nodeLogger: newNodeLogger(node),
}
}
func (e *executeSuccessNode) Run(ctx context.Context) error {
e.AddOutput(ctx,
e.node.Name,
"进入执行成功分支",
)
func (n *executeSuccessNode) Process(ctx context.Context) error {
// 此类型节点不需要执行任何操作，直接返回
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "进入执行成功分支")
return nil
}

22
internal/workflow/node-processor/notify_node.go
View File

@@ -9,44 +9,46 @@ import (
)
type notifyNode struct {
node *domain.WorkflowNode
settingsRepo settingsRepository
node *domain.WorkflowNode
*nodeLogger
settingsRepo settingsRepository
}
func NewNotifyNode(node *domain.WorkflowNode) *notifyNode {
return &notifyNode{
node: node,
nodeLogger: NewNodeLogger(node),
node: node,
nodeLogger: newNodeLogger(node),
settingsRepo: repository.NewSettingsRepository(),
}
}
func (n *notifyNode) Run(ctx context.Context) error {
n.AddOutput(ctx, n.node.Name, "开始执行")
func (n *notifyNode) Process(ctx context.Context) error {
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "进入推送通知节点")
nodeConfig := n.node.GetConfigForNotify()
// 获取通知配置
settings, err := n.settingsRepo.GetByName(ctx, "notifyChannels")
if err != nil {
n.AddOutput(ctx, n.node.Name, "获取通知配置失败", err.Error())
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "获取通知配置失败", err.Error())
return err
}
// 获取通知渠道
channelConfig, err := settings.GetNotifyChannelConfig(nodeConfig.Channel)
if err != nil {
n.AddOutput(ctx, n.node.Name, "获取通知渠道配置失败", err.Error())
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "获取通知渠道配置失败", err.Error())
return err
}
// 发送通知
if err := notify.SendToChannel(nodeConfig.Subject, nodeConfig.Message, nodeConfig.Channel, channelConfig); err != nil {
n.AddOutput(ctx, n.node.Name, "发送通知失败", err.Error())
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "发送通知失败", err.Error())
return err
}
n.AddOutput(ctx, n.node.Name, "发送通知成功")
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "发送通知成功")
return nil
}

33
internal/workflow/node-processor/processor.go
View File

@@ -9,9 +9,10 @@ import (
)
type NodeProcessor interface {
Run(ctx context.Context) error
Log(ctx context.Context) *domain.WorkflowRunLog
AddOutput(ctx context.Context, title, content string, err ...string)
Process(ctx context.Context) error
GetLog(ctx context.Context) *domain.WorkflowRunLog
AppendLogRecord(ctx context.Context, level domain.WorkflowRunLogLevel, content string, err ...string)
}
type nodeLogger struct {
@@ -23,39 +24,41 @@ type certificateRepository interface {
}
type workflowOutputRepository interface {
GetByNodeId(ctx context.Context, nodeId string) (*domain.WorkflowOutput, error)
Save(ctx context.Context, output *domain.WorkflowOutput, certificate *domain.Certificate, cb func(id string) error) error
GetByNodeId(ctx context.Context, workflowNodeId string) (*domain.WorkflowOutput, error)
Save(ctx context.Context, workflowOutput *domain.WorkflowOutput) (*domain.WorkflowOutput, error)
SaveWithCertificate(ctx context.Context, workflowOutput *domain.WorkflowOutput, certificate *domain.Certificate) (*domain.WorkflowOutput, error)
}
type settingsRepository interface {
GetByName(ctx context.Context, name string) (*domain.Settings, error)
}
func NewNodeLogger(node *domain.WorkflowNode) *nodeLogger {
func newNodeLogger(node *domain.WorkflowNode) *nodeLogger {
return &nodeLogger{
log: &domain.WorkflowRunLog{
NodeId: node.Id,
NodeName: node.Name,
Outputs: make([]domain.WorkflowRunLogOutput, 0),
Records: make([]domain.WorkflowRunLogRecord, 0),
},
}
}
func (l *nodeLogger) Log(ctx context.Context) *domain.WorkflowRunLog {
func (l *nodeLogger) GetLog(ctx context.Context) *domain.WorkflowRunLog {
return l.log
}
func (l *nodeLogger) AddOutput(ctx context.Context, title, content string, err ...string) {
output := domain.WorkflowRunLogOutput{
func (l *nodeLogger) AppendLogRecord(ctx context.Context, level domain.WorkflowRunLogLevel, content string, err ...string) {
record := domain.WorkflowRunLogRecord{
Time: time.Now().UTC().Format(time.RFC3339),
Title: title,
Level: level,
Content: content,
}
if len(err) > 0 {
output.Error = err[0]
record.Error = err[0]
l.log.Error = err[0]
}
l.log.Outputs = append(l.log.Outputs, output)
l.log.Records = append(l.log.Records, record)
}
func GetProcessor(node *domain.WorkflowNode) (NodeProcessor, error) {
@@ -83,3 +86,7 @@ func GetProcessor(node *domain.WorkflowNode) (NodeProcessor, error) {
func getContextWorkflowId(ctx context.Context) string {
return ctx.Value("workflow_id").(string)
}
func getContextWorkflowRunId(ctx context.Context) string {
return ctx.Value("workflow_run_id").(string)
}

8
internal/workflow/node-processor/start_node.go
View File

@@ -14,13 +14,13 @@ type startNode struct {
func NewStartNode(node *domain.WorkflowNode) *startNode {
return &startNode{
node: node,
nodeLogger: NewNodeLogger(node),
nodeLogger: newNodeLogger(node),
}
}
func (s *startNode) Run(ctx context.Context) error {
// 开始节点没有任何操作
s.AddOutput(ctx, s.node.Name, "完成")
func (n *startNode) Process(ctx context.Context) error {
// 此类型节点不需要执行任何操作，直接返回
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "进入开始节点")
return nil
}

110
internal/workflow/node-processor/upload_node.go
View File

@@ -12,90 +12,94 @@ import (
)
type uploadNode struct {
node *domain.WorkflowNode
outputRepo workflowOutputRepository
node *domain.WorkflowNode
*nodeLogger
certRepo certificateRepository
outputRepo workflowOutputRepository
}
func NewUploadNode(node *domain.WorkflowNode) *uploadNode {
return &uploadNode{
node: node,
nodeLogger: NewNodeLogger(node),
nodeLogger: newNodeLogger(node),
certRepo: repository.NewCertificateRepository(),
outputRepo: repository.NewWorkflowOutputRepository(),
}
}
// Run 上传证书节点执行
// 包含上传证书的工作流，理论上应该手动执行，如果每天定时执行，也只是重新保存一下
func (n *uploadNode) Run(ctx context.Context) error {
n.AddOutput(ctx,
n.node.Name,
"进入上传证书节点",
)
func (n *uploadNode) Process(ctx context.Context) error {
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "进入上传证书节点")
config := n.node.GetConfigForUpload()
nodeConfig := n.node.GetConfigForUpload()
// 检查证书是否过期
// 如果证书过期，则直接返回错误
certX509, err := certs.ParseCertificateFromPEM(config.Certificate)
if err != nil {
n.AddOutput(ctx,
n.node.Name,
"解析证书失败",
)
// 查询上次执行结果
lastOutput, err := n.outputRepo.GetByNodeId(ctx, n.node.Id)
if err != nil && !domain.IsRecordNotFoundError(err) {
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "查询申请记录失败", err.Error())
return err
}
// 检测是否可以跳过本次执行
if skippable, skipReason := n.checkCanSkip(ctx, lastOutput); skippable {
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, skipReason)
return nil
}
// 检查证书是否过期
// 如果证书过期，则直接返回错误
certX509, err := certs.ParseCertificateFromPEM(nodeConfig.Certificate)
if err != nil {
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "解析证书失败")
return err
}
if time.Now().After(certX509.NotAfter) {
n.AddOutput(ctx,
n.node.Name,
"证书已过期",
)
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelWarn, "证书已过期")
return errors.New("certificate is expired")
}
// 生成证书实体
certificate := &domain.Certificate{
Source: domain.CertificateSourceTypeUpload,
SubjectAltNames: strings.Join(certX509.DNSNames, ";"),
Certificate: config.Certificate,
PrivateKey: config.PrivateKey,
EffectAt: certX509.NotBefore,
ExpireAt: certX509.NotAfter,
WorkflowId: getContextWorkflowId(ctx),
WorkflowNodeId: n.node.Id,
Source: domain.CertificateSourceTypeUpload,
}
certificate.PopulateFromPEM(nodeConfig.Certificate, nodeConfig.PrivateKey)
// 保存执行结果
// TODO: 先保持一个节点始终只有一个输出，后续增加版本控制
currentOutput := &domain.WorkflowOutput{
output := &domain.WorkflowOutput{
WorkflowId: getContextWorkflowId(ctx),
RunId: getContextWorkflowRunId(ctx),
NodeId: n.node.Id,
Node: n.node,
Succeeded: true,
Outputs: n.node.Outputs,
}
// 查询上次执行结果
lastOutput, err := n.outputRepo.GetByNodeId(ctx, n.node.Id)
if err != nil && !domain.IsRecordNotFoundError(err) {
n.AddOutput(ctx, n.node.Name, "查询上传记录失败", err.Error())
if _, err := n.outputRepo.SaveWithCertificate(ctx, output, certificate); err != nil {
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelError, "保存上传记录失败", err.Error())
return err
}
if lastOutput != nil {
currentOutput.Id = lastOutput.Id
}
if err := n.outputRepo.Save(ctx, currentOutput, certificate, func(id string) error {
if certificate != nil {
certificate.WorkflowOutputId = id
}
return nil
}); err != nil {
n.AddOutput(ctx, n.node.Name, "保存上传记录失败", err.Error())
return err
}
n.AddOutput(ctx, n.node.Name, "保存上传记录成功")
n.AppendLogRecord(ctx, domain.WorkflowRunLogLevelInfo, "保存上传记录成功")
return nil
}
func (n *uploadNode) checkCanSkip(ctx context.Context, lastOutput *domain.WorkflowOutput) (skip bool, reason string) {
if lastOutput != nil && lastOutput.Succeeded {
// 比较和上次上传时的关键配置（即影响证书上传的）参数是否一致
currentNodeConfig := n.node.GetConfigForUpload()
lastNodeConfig := lastOutput.Node.GetConfigForUpload()
if strings.TrimSpace(currentNodeConfig.Certificate) != strings.TrimSpace(lastNodeConfig.Certificate) {
return false, "配置项变化：证书"
}
if strings.TrimSpace(currentNodeConfig.PrivateKey) != strings.TrimSpace(lastNodeConfig.PrivateKey) {
return false, "配置项变化：私钥"
}
lastCertificate, _ := n.certRepo.GetByWorkflowNodeId(ctx, n.node.Id)
if lastCertificate != nil {
return true, "已上传过证书"
}
}
return false, ""
}

89
internal/workflow/processor/processor.go
View File

@@ -1,89 +0,0 @@
package processor
import (
"context"
"github.com/usual2970/certimate/internal/domain"
nodes "github.com/usual2970/certimate/internal/workflow/node-processor"
)
type workflowProcessor struct {
workflow *domain.Workflow
logs []domain.WorkflowRunLog
}
func NewWorkflowProcessor(workflow *domain.Workflow) *workflowProcessor {
return &workflowProcessor{
workflow: workflow,
logs: make([]domain.WorkflowRunLog, 0),
}
}
func (w *workflowProcessor) Run(ctx context.Context) error {
ctx = setContextWorkflowId(ctx, w.workflow.Id)
return w.processNode(ctx, w.workflow.Content)
}
func (w *workflowProcessor) GetRunLogs() []domain.WorkflowRunLog {
return w.logs
}
func (w *workflowProcessor) processNode(ctx context.Context, node *domain.WorkflowNode) error {
current := node
for current != nil {
if current.Type == domain.WorkflowNodeTypeBranch || current.Type == domain.WorkflowNodeTypeExecuteResultBranch {
for _, branch := range current.Branches {
if err := w.processNode(ctx, &branch); err != nil {
continue
}
}
}
var processor nodes.NodeProcessor
var runErr error
for {
if current.Type != domain.WorkflowNodeTypeBranch && current.Type != domain.WorkflowNodeTypeExecuteResultBranch {
processor, runErr = nodes.GetProcessor(current)
if runErr != nil {
break
}
runErr = processor.Run(ctx)
log := processor.Log(ctx)
if log != nil {
w.logs = append(w.logs, *log)
}
if runErr != nil {
break
}
}
break
}
if runErr != nil && current.Next != nil && current.Next.Type != domain.WorkflowNodeTypeExecuteResultBranch {
return runErr
} else if runErr != nil && current.Next != nil && current.Next.Type == domain.WorkflowNodeTypeExecuteResultBranch {
current = getBranchByType(current.Next.Branches, domain.WorkflowNodeTypeExecuteFailure)
} else if runErr == nil && current.Next != nil && current.Next.Type == domain.WorkflowNodeTypeExecuteResultBranch {
current = getBranchByType(current.Next.Branches, domain.WorkflowNodeTypeExecuteSuccess)
} else {
current = current.Next
}
}
return nil
}
func setContextWorkflowId(ctx context.Context, id string) context.Context {
return context.WithValue(ctx, "workflow_id", id)
}
func getBranchByType(branches []domain.WorkflowNode, nodeType domain.WorkflowNodeType) *domain.WorkflowNode {
for _, branch := range branches {
if branch.Type == nodeType {
return &branch
}
}
return nil
}

165
internal/workflow/service.go
View File

@@ -4,70 +4,64 @@ import (
"context"
"errors"
"fmt"
"sync"
"time"
"github.com/usual2970/certimate/internal/app"
"github.com/usual2970/certimate/internal/domain"
"github.com/usual2970/certimate/internal/domain/dtos"
processor "github.com/usual2970/certimate/internal/workflow/processor"
"github.com/usual2970/certimate/internal/workflow/dispatcher"
)
const defaultRoutines = 10
type workflowRunData struct {
Workflow *domain.Workflow
RunTrigger domain.WorkflowTriggerType
}
type workflowRepository interface {
ListEnabledAuto(ctx context.Context) ([]*domain.Workflow, error)
GetById(ctx context.Context, id string) (*domain.Workflow, error)
Save(ctx context.Context, workflow *domain.Workflow) (*domain.Workflow, error)
SaveRun(ctx context.Context, workflowRun *domain.WorkflowRun) (*domain.WorkflowRun, error)
}
type workflowRunRepository interface {
GetById(ctx context.Context, id string) (*domain.WorkflowRun, error)
Save(ctx context.Context, workflowRun *domain.WorkflowRun) (*domain.WorkflowRun, error)
}
type WorkflowService struct {
ch chan *workflowRunData
repo workflowRepository
wg sync.WaitGroup
cancel context.CancelFunc
dispatcher *dispatcher.WorkflowDispatcher
workflowRepo workflowRepository
workflowRunRepo workflowRunRepository
}
func NewWorkflowService(repo workflowRepository) *WorkflowService {
func NewWorkflowService(workflowRepo workflowRepository, workflowRunRepo workflowRunRepository) *WorkflowService {
srv := &WorkflowService{
repo: repo,
ch: make(chan *workflowRunData, 1),
dispatcher: dispatcher.GetSingletonDispatcher(workflowRepo, workflowRunRepo),
workflowRepo: workflowRepo,
workflowRunRepo: workflowRunRepo,
}
ctx, cancel := context.WithCancel(context.Background())
srv.cancel = cancel
srv.wg.Add(defaultRoutines)
for i := 0; i < defaultRoutines; i++ {
go srv.run(ctx)
}
return srv
}
func (s *WorkflowService) InitSchedule(ctx context.Context) error {
workflows, err := s.repo.ListEnabledAuto(ctx)
workflows, err := s.workflowRepo.ListEnabledAuto(ctx)
if err != nil {
return err
}
scheduler := app.GetScheduler()
for _, workflow := range workflows {
var errs []error
err := scheduler.Add(fmt.Sprintf("workflow#%s", workflow.Id), workflow.TriggerCron, func() {
s.StartRun(ctx, &dtos.WorkflowStartRunReq{
WorkflowId: workflow.Id,
Trigger: domain.WorkflowTriggerTypeAuto,
RunTrigger: domain.WorkflowTriggerTypeAuto,
})
})
if err != nil {
app.GetLogger().Error("failed to add schedule", "err", err)
return err
errs = append(errs, err)
}
if len(errs) > 0 {
return errors.Join(errs...)
}
}
@@ -75,97 +69,56 @@ func (s *WorkflowService) InitSchedule(ctx context.Context) error {
}
func (s *WorkflowService) StartRun(ctx context.Context, req *dtos.WorkflowStartRunReq) error {
workflow, err := s.repo.GetById(ctx, req.WorkflowId)
workflow, err := s.workflowRepo.GetById(ctx, req.WorkflowId)
if err != nil {
app.GetLogger().Error("failed to get workflow", "id", req.WorkflowId, "err", err)
return err
}
if workflow.LastRunStatus == domain.WorkflowRunStatusTypeRunning {
return errors.New("workflow is running")
if workflow.LastRunStatus == domain.WorkflowRunStatusTypePending || workflow.LastRunStatus == domain.WorkflowRunStatusTypeRunning {
return errors.New("workflow is already pending or running")
}
workflow.LastRunTime = time.Now()
workflow.LastRunStatus = domain.WorkflowRunStatusTypePending
workflow.LastRunId = ""
if resp, err := s.repo.Save(ctx, workflow); err != nil {
return err
} else {
workflow = resp
}
s.ch <- &workflowRunData{
Workflow: workflow,
RunTrigger: req.Trigger,
}
return nil
}
func (s *WorkflowService) CancelRun(ctx context.Context, req *dtos.WorkflowCancelRunReq) error {
// TODO: 取消运行，防止因为某些原因意外挂起（如进程被杀死）导致工作流一直处于 running 状态无法重新运行
return errors.New("TODO: 尚未实现")
}
func (s *WorkflowService) Stop(ctx context.Context) {
s.cancel()
s.wg.Wait()
}
func (s *WorkflowService) run(ctx context.Context) {
defer s.wg.Done()
for {
select {
case data := <-s.ch:
if err := s.runWithData(ctx, data); err != nil {
app.GetLogger().Error("failed to run workflow", "id", data.Workflow.Id, "err", err)
}
case <-ctx.Done():
return
}
}
}
func (s *WorkflowService) runWithData(ctx context.Context, runData *workflowRunData) error {
workflow := runData.Workflow
run := &domain.WorkflowRun{
WorkflowId: workflow.Id,
Status: domain.WorkflowRunStatusTypeRunning,
Trigger: runData.RunTrigger,
Status: domain.WorkflowRunStatusTypePending,
Trigger: req.RunTrigger,
StartedAt: time.Now(),
}
if resp, err := s.repo.SaveRun(ctx, run); err != nil {
if resp, err := s.workflowRunRepo.Save(ctx, run); err != nil {
return err
} else {
run = resp
}
processor := processor.NewWorkflowProcessor(workflow)
if runErr := processor.Run(ctx); runErr != nil {
run.Status = domain.WorkflowRunStatusTypeFailed
run.EndedAt = time.Now()
run.Logs = processor.GetRunLogs()
run.Error = runErr.Error()
if _, err := s.repo.SaveRun(ctx, run); err != nil {
app.GetLogger().Error("failed to save workflow run", "err", err)
}
return fmt.Errorf("failed to run workflow: %w", runErr)
}
run.EndedAt = time.Now()
run.Logs = processor.GetRunLogs()
run.Error = domain.WorkflowRunLogs(run.Logs).ErrorString()
if run.Error == "" {
run.Status = domain.WorkflowRunStatusTypeSucceeded
} else {
run.Status = domain.WorkflowRunStatusTypeFailed
}
if _, err := s.repo.SaveRun(ctx, run); err != nil {
app.GetLogger().Error("failed to save workflow run", "err", err)
return err
}
s.dispatcher.Dispatch(&dispatcher.WorkflowWorkerData{
WorkflowId: workflow.Id,
WorkflowContent: workflow.Content,
RunId: run.Id,
})
return nil
}
func (s *WorkflowService) CancelRun(ctx context.Context, req *dtos.WorkflowCancelRunReq) error {
workflow, err := s.workflowRepo.GetById(ctx, req.WorkflowId)
if err != nil {
return err
}
workflowRun, err := s.workflowRunRepo.GetById(ctx, req.RunId)
if err != nil {
return err
} else if workflowRun.WorkflowId != workflow.Id {
return errors.New("workflow run not found")
} else if workflowRun.Status != domain.WorkflowRunStatusTypePending && workflowRun.Status != domain.WorkflowRunStatusTypeRunning {
return errors.New("workflow run is not pending or running")
}
s.dispatcher.Cancel(workflowRun.Id)
return nil
}
func (s *WorkflowService) Shutdown(ctx context.Context) {
s.dispatcher.Shutdown()
}